2024年最全Go操作RabbitMQ

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加戳这里获取

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

err = r.conn.Close()
if err != nil {
    return
}

}

// 错误处理函数

func (r *RabbitMQ) failOnErr(err error, message string) {

if err != nil {

log.Fatalf(“%s:%s”, message, err)

}

// NewRabbitMQSimple 创建简单模式下RabbitMQ实例

func NewRabbitMQSimple(queueName string) *RabbitMQ {

//创建RabbitMQ实例

rabbitmq := NewRabbitMQ(queueName, “”, “”)

var err error

//获取connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnErr(err, “failed to connect rabb”+

“itmq!”)

//获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnErr(err, “failed to open a channel”)

return rabbitmq

}

// PublishSimple simple模式队列生产

func (r *RabbitMQ) PublishSimple(message string) {

//1.申请队列，如果队列不存在会自动创建，存在则跳过创建

_, err := r.channel.QueueDeclare(

// queue:队列名称

r.QueueName,

//durable:是否持久化，当mq重启之后，还在

true,

//exclusive:是否独占即只能有一个消费者监听这个队列

false,

//autoDelete:是否自动删除。当没有Consumer时，自动删除掉

false,

//noWait:是否阻塞处理。true:不阻塞，false:阻塞

false,

//arguments:其他属性

nil,

)

if err != nil {

fmt.Println(err)

}

//调用channel 发送消息到队列中

r.channel.Publish(

r.Exchange,

r.QueueName,

//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者

false,

//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者

false,

amqp.Publishing{

ContentType: “text/plain”,

Body: []byte(message),

})

}

// ConsumeSimple simple模式下消费者

func (r *RabbitMQ) ConsumeSimple() {

//1.申请队列，如果队列不存在会自动创建，存在则跳过创建

q, err := r.channel.QueueDeclare(

//队列名称

r.QueueName,

//是否持久化

true,

//是否自动删除

false,

//是否具有排他性

false,

//是否阻塞处理

false,

//额外的属性

nil,

)

if err != nil {

fmt.Println(err)

}

//接收消息

msgs, err := r.channel.Consume(

q.Name, // queue

//用来区分多个消费者

“”, // consumer

//是否自动应答

true, // auto-ack

//是否独有

false, // exclusive

//设置为true，表示不能将同一个Conenction中生产者发送的消息传递给这个Connection中的消费者

false, // no-local

//列是否阻塞

false, // no-wait

nil, // args

)

if err != nil {

fmt.Println(err)

}

forever := make(chan bool)
//启用协程处理消息
go func() {
    for d := range msgs {
        //消息逻辑处理，可以自行设计逻辑
        log.Printf("Received a message: %s", d.Body)
    }
}()
log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever

}

##### 1.2.2. SimplePublish.go

package main

import (

“fmt”

“qiutian.com/study/02.rabbitmq/Simple/mq”

)

func main() {

rabbitmq := mq.NewRabbitMQSimple(“hello.queue1”)

rabbitmq.PublishSimple(“Hello akita!”)

fmt.Println(“发送成功！”)

}

##### 1.2.3. mainSimpleRecieve.go

package main

import (

“qiutian.com/study/02.rabbitmq/Simple/mq”

)

// simple模式消费者

func main() {

// queueName 替换成你自己的队列名

rabbitmq := mq.NewRabbitMQSimple(“hello.queue1”)

rabbitmq.ConsumeSimple()

}

### 2. 工作模式（Work）
![](https://img-blog.csdnimg.cn/img_convert/b3fb23f3054a68c0bbc4f907b6188136.png)
* **特点**：
* + 多个消费者同时从同一个队列中获取消息并处理，每个消息只能被一个消费者处理。
	+ 消息在被消费者处理后会从队列中移除。
* **真实应用场景**：
* + 任务分发系统：多个消费者同时从队列中获取任务，每个消费者处理一个任务，以实现任务的负载均衡。
* **优点**：
* + 可以通过增加消费者来提高消息处理的并发性。
* **缺点**：
* + 没有消息持久化，一旦 RabbitMQ 服务器重启，队列中的消息将会丢失。
#### 2.1. 目录结构
Work
-mq
–-rabbitmq.go //这个是RabbitMQ的封装
-WorkPublish
–-mainWorkPublish.go //生产者发送消息
-WorkRecieve1
–-mainWorkRecieve1.go // 消费者接受消息
-WorkRecieve2
–-mainWorkRecieve2.go // 消费者接受消息
![](https://img-blog.csdnimg.cn/img_convert/e6971835f2cd917c41c57d03503f5152.png)
#### 2.2. 代码实现
##### 2.2.1. rabbitmq.go

package mq

import (

“fmt”

“log”

"github.com/streadway/amqp"

)

// MQURL amqp://user:password@host:port/vhost

// amqp://是固定参数，这个信息是固定不变的。后面两个是用户名密码ip地址端口号Virtual Host

// 如果vhost是“/”就输入/%2F，/%2F代表斜杠

const MQURL = “amqp://test:123123@127.0.0.1:5672/%2F”

// RabbitMQ rabbitMQ结构体

type RabbitMQ struct {

conn *amqp.Connection

channel *amqp.Channel

//队列名称

QueueName string

//交换机名称

Exchange string

//bind Key 名称

Key string

//连接信息

Mqurl string

}

// NewRabbitMQ 创建结构体实例

func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {

return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}

}

// Destory 断开channel 和 connection

func (r *RabbitMQ) Destory() {

r.channel.Close()

r.conn.Close()

}

// 错误处理函数

func (r *RabbitMQ) failOnErr(err error, message string) {

if err != nil {

log.Fatalf(“%s:%s”, message, err)

panic(fmt.Sprintf(“%s:%s”, message, err))

}

// NewRabbitMQWork 创建work模式下RabbitMQ实例

func NewRabbitMQWork(queueName string) *RabbitMQ {

//创建RabbitMQ实例

rabbitmq := NewRabbitMQ(queueName, “”, “”)

var err error

//获取connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnErr(err, “failed to connect rabb”+

“itmq!”)

//获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnErr(err, “failed to open a channel”)

return rabbitmq

}

// PublishWork work模式队列生产

func (r *RabbitMQ) PublishWork(message string) {

//1.申请队列，如果队列不存在会自动创建，存在则跳过创建

_, err := r.channel.QueueDeclare(

// queue:队列名称

r.QueueName,

//durable:是否持久化，当mq重启之后，还在

true,

//exclusive:是否独占即只能有一个消费者监听这个队列

false,

//autoDelete:是否自动删除。当没有Consumer时，自动删除掉

false,

//noWait:是否阻塞处理。true:不阻塞，false:阻塞

false,

//arguments:其他属性

nil,

)

if err != nil {

fmt.Println(err)

}

//调用channel 发送消息到队列中

r.channel.Publish(

r.Exchange,

r.QueueName,

//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者

false,

//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者

false,

amqp.Publishing{

ContentType: “text/plain”,

Body: []byte(message),

})

}

// ConsumeWork work 模式下消费者

func (r *RabbitMQ) ConsumeWork() {

//1.申请队列，如果队列不存在会自动创建，存在则跳过创建

q, err := r.channel.QueueDeclare(

//队列名称

r.QueueName,

//是否持久化

true,

//是否自动删除

false,

//是否具有排他性

false,

//是否阻塞处理

false,

//额外的属性

nil,

)

if err != nil {

fmt.Println(err)

}

//接收消息

msgs, err := r.channel.Consume(

q.Name, // queue

//用来区分多个消费者

“”, // consumer

//是否自动应答

true, // auto-ack

//是否独有

false, // exclusive

//设置为true，表示不能将同一个Conenction中生产者发送的消息传递给这个Connection中的消费者

false, // no-local

//列是否阻塞

false, // no-wait

nil, // args

)

if err != nil {

fmt.Println(err)

}

forever := make(chan bool)
//启用协程处理消息
go func() {
    for d := range msgs {
        //消息逻辑处理，可以自行设计逻辑
        log.Printf("Received a message: %s", d.Body)
    }
}()
log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever

}

##### 2.2.2. mainWorkPublish.go

package main

import (

“fmt”

“qiutian.com/study/02.rabbitmq/Work/mq”

“strconv”

“time”

)

func main() {

rabbitmq := mq.NewRabbitMQWork(“hello.queue1”)

for i := 0; i <= 100; i++ {

rabbitmq.PublishWork(“Hello akita!” + strconv.Itoa(i))

time.Sleep(1 * time.Second)

fmt.Println(i)

}

##### 2.2.3. mainWorkRecieve1.go

package main

import (

“qiutian.com/study/02.rabbitmq/Work/mq”

)

func main() {

// queueName 替换成你自己的队列名

rabbitmq := mq.NewRabbitMQWork(“hello.queue1”)

rabbitmq.ConsumeWork()

}

##### 2.2.4. mainWorkRecieve2.go

package main

import (

“qiutian.com/study/02.rabbitmq/Work/mq”

)

func main() {

// queueName 替换成你自己的队列名

rabbitmq := mq.NewRabbitMQWork(“hello.queue1”)

rabbitmq.ConsumeWork()

}

### 3. 订阅模式(Publish)
TIPS:
在订阅消费者模式下，消费者后监听到的数据取决于它开始监听的时间。如果消费者在生产者生产数据之前开始监听，那么它将不会消费到生产者在开始监听之前发布的数据。只有在消费者开始监听之后，生产者发布的数据才会被消费者接收和处理。因此，如果你想让消费者消费之前发布的数据，你需要确保消费者在生产者发布数据之后开始监听。
![](https://img-blog.csdnimg.cn/img_convert/5a3b9ab6eff8ed7748f29fb259953e4c.png)
* **特点**：
* + 生产者将消息发送到一个交换机中，交换机将消息广播给所有绑定了该交换机的队列。
	+ 每个消息都会被交换机广播给所有绑定了该交换机的队列。
* **真实应用场景**：
* + 新闻推送系统：生产者将最新的新闻发布到交换机中，每个订阅者（消费者）都会收到相同的新闻。
* **优点**：
* + 实现了消息的广播功能，适用于多个消费者同时接收相同消息的场景。
* **缺点**：
* + 无法根据消费者的兴趣或条件选择性地接收消息。
#### 3.1. 目录结构
![](https://img-blog.csdnimg.cn/img_convert/95cb5567f76ee1082e7c288c5b01794a.png)
#### 3.2. 代码实现
###### 3.2.1.1. rabbitmq.go

package mq

import (

“fmt”

“log”

"github.com/streadway/amqp"

)

// MQURL amqp://user:password@host:port/vhost

// amqp://是固定参数，这个信息是固定不变的。后面两个是用户名密码ip地址端口号Virtual Host

// 如果vhost是“/”就输入/%2F，/%2F代表斜杠

const MQURL = “amqp://test:123123@127.0.0.1:5672/%2F”

// RabbitMQ rabbitMQ结构体

type RabbitMQ struct {

conn *amqp.Connection

channel *amqp.Channel

//队列名称

QueueName string

//交换机名称

Exchange string

//bind Key 名称

Key string

//连接信息

Mqurl string

}

// NewRabbitMQ 创建结构体实例

func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {

return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}

}

// Destory 断开channel 和 connection

func (r *RabbitMQ) Destory() {

r.channel.Close()

r.conn.Close()

}

// 错误处理函数

func (r *RabbitMQ) failOnErr(err error, message string) {

if err != nil {

log.Fatalf(“%s:%s”, message, err)

panic(fmt.Sprintf(“%s:%s”, message, err))

}

// NewRabbitMQPub 创建Pub模式下RabbitMQ实例

func NewRabbitMQPub(exchangeName string) *RabbitMQ {

//创建RabbitMQ实例

rabbitmq := NewRabbitMQ(“”, exchangeName, “”)

var err error

//获取connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnErr(err, “failed to connect rabb”+

“itmq!”)

//获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnErr(err, “failed to open a channel”)

return rabbitmq

}

// PublishPub Pub模式队列生产

func (r *RabbitMQ) PublishPub(message string) {

//1.尝试创建交换机

err := r.channel.ExchangeDeclare(

//交换机名称

r.Exchange,

//交换机类型

“fanout”,

//是否持久化

true,

//是否自动删除

false,

//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定

false,

//是否阻塞处理

false,

//其他属性

nil,

)

r.failOnErr(err, “Failed to declare an exchange”)

//2.发送消息
//发送消息
//参数：
//1.交换机
//2.队列，
//3.是否强制性路由到队列，如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还
//给发送者
//4.是否立即发送
//5.其他属性
//6.发送消息的内容
//调用channel 发送消息到队列中
r.channel.Publish(
    r.Exchange,
    "",
    //如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
    false,
    //如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
    false,
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(message),
    })

}

// ConsumePub pub 模式下消费者

func (r *RabbitMQ) ConsumePub() {

//1.试探性创建交换机

err := r.channel.ExchangeDeclare(

// 交换机名称

r.Exchange,

//交换机类型

“fanout”,

//是否持久化

true,

// 是否自动删除

false,

//YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定

false,

nil,

)

r.failOnErr(err, “Failed to declare an exchange”)

//2.试探性创建队列，这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
    //随机生产队列名称
    "",
    //是否持久化
    false,
    //是否自动删除
    false,
    //是否具有排他性
    true,
    //是否阻塞处理
    false,
    //其他参数
    nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
    q.Name,
    //在pub/sub模式下，这里的key要为空
    "",
    r.Exchange,
    false,
    nil)
//消费消息
messages, err := r.channel.Consume(
    q.Name,
    "",
    true,
    false,
    false,
    false,
    nil,
)
forever := make(chan bool)
//启用协程处理消息
go func() {
    for d := range messages {
        log.Printf("Received a message: %s", d.Body)
    }
}()
fmt.Println("退出请按 CTRL+C\n")
<-forever

}

###### 3.2.1.2. mainPubPublish.go

package main

import (

“fmt”

“qiutian.com/study/02.rabbitmq/Pub/mq”

“strconv”

“time”

)

func main() {

rabbitmq := mq.NewRabbitMQPub(“newExchange”)

for i := 0; i < 100; i++ {

rabbitmq.PublishPub(“订阅模式生产第” +

strconv.Itoa(i) + “条” + “数据”)

fmt.Println(“订阅模式生产第” +

strconv.Itoa(i) + “条” + “数据”)

time.Sleep(1 * time.Second)

}

###### 3.2.1.3. mainPubRecieve1.go

package main

import (

“qiutian.com/study/02.rabbitmq/Pub/mq”

)

func main() {

rabbitmq := mq.NewRabbitMQPub(“” +

“newExchange”)

rabbitmq.ConsumePub()

}

###### 3.2.1.4. mainPubRecieve2.go

package main

import (

“qiutian.com/study/02.rabbitmq/Pub/mq”

)

func main() {

rabbitmq := mq.NewRabbitMQPub(“” +

“newExchange”)

rabbitmq.ConsumePub()

}

### 4. 路由模式（routing）
![](https://img-blog.csdnimg.cn/img_convert/967b8a6351bacdf4a1f7d1a31f2f21e8.png)
路由模式（Routing Mode）是消息队列中一种常见的消息路由模式，它允许消息的发送者（Producer）将消息发送到指定的交换机（Exchange），而交换机则根据消息的路由键（Routing Key）将消息路由到符合条件的队列（Queue）。路由模式提供了一种灵活的消息路由机制，使得消息的分发可以根据不同的路由规则进行定制。
**优点：**
1. **灵活性高：** 路由模式允许根据消息的属性或条件将消息路由到不同的队列或消费者，从而实现灵活的消息处理和路由策略。
2. **解耦性强：** 路由模式可以将消息的生产者和消费者解耦，使它们之间相互独立。生产者只需将消息发送到交换机，而不需要知道具体的消费者是谁，消费者也只需订阅感兴趣的消息，而不需要知道消息的生产者是谁，从而实现了系统的解耦。
3. **灵活的消息过滤和路由：** 路由模式允许根据消息的属性、标签或主题将消息路由到不同的队列或消费者，实现精确的消息过滤和路由，从而提高了系统的灵活性和可扩展性。
4. **支持多种路由策略：** 路由模式支持多种路由策略，如直连路由、主题路由、分发路由等，可以根据不同的业务需求选择合适的路由策略，从而满足不同的应用场景。
**缺点：**
1. **配置复杂：** 路由模式的配置相对复杂，需要定义交换机、队列和绑定关系，以及配置路由规则等，如果配置不当可能会导致消息路由错误或丢失。
2. **性能损耗：** 路由模式需要对消息进行额外的路由和过滤操作，可能会增加系统的消息处理延迟和性能消耗，特别是在消息量较大的情况下。
3. **消息堆积风险：** 如果路由模式配置不当或消息处理能力不足，可能会导致消息堆积的风险，从而影响系统的稳定性和可靠性。
4. **维护成本高：** 路由模式需要维护交换机、队列和绑定关系等配置信息，特别是在系统规模较大或消息路由策略较复杂的情况下，维护成本可能会较高。
**使用场景：**
1. **微服务架构下的消息路由**：在微服务架构中，不同的服务可能需要处理不同类型的消息。通过路由模式，可以将消息根据服务的需求进行路由，确保每个服务只接收到与其相关的消息，从而实现解耦和灵活性。
2. **实时监控与日志分析**：在监控系统或日志分析系统中，需要根据不同的指标或日志级别对数据进行分类和分析。路由模式可以根据数据的属性或标签将数据路由到不同的处理节点，以实现实时监控和日志分析。
3. **任务调度与负载均衡**：在任务调度系统中，任务可能具有不同的优先级或类型，需要根据任务的属性进行分发和调度。通过路由模式，可以将任务路由到不同的工作节点，实现任务的负载均衡和优先级调度。
4. **多级消息过滤与订阅**：在订阅发布系统中，订阅者可能对不同类型或主题的消息感兴趣。路由模式可以根据消息的属性或主题将消息路由到不同的订阅者，实现精确的消息过滤和订阅。
5. **数据同步与复制**：在分布式系统中，可能需要将数据复制到不同的节点或数据中心，以实现数据的备份和灾备。通过路由模式，可以将数据根据不同的条件进行路由和复制，确保数据的一致性和可靠性。
#### 4.1. 目录结构
![](https://img-blog.csdnimg.cn/img_convert/838c922fdd5e15664020b3b85ff5b66f.png)
#### 4.2. 代码实现
##### 4.2.1. rabbitmq.go

package mq

import (

“fmt”

“log”

"github.com/streadway/amqp"

)

// MQURL amqp://user:password@host:port/vhost

// amqp://是固定参数，这个信息是固定不变的。后面两个是用户名密码ip地址端口号Virtual Host

// 如果vhost是“/”就输入/%2F，/%2F代表斜杠

const MQURL = “amqp://test:123123@127.0.0.1:5672/%2F”

// RabbitMQ rabbitMQ结构体

type RabbitMQ struct {

conn *amqp.Connection

channel *amqp.Channel

//队列名称

QueueName string

//交换机名称

Exchange string

//bind Key 名称

Key string

//连接信息

Mqurl string

}

// NewRabbitMQ 创建结构体实例

func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {

return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}

}

// Destory 断开channel 和 connection

func (r *RabbitMQ) Destory() {

r.channel.Close()

r.conn.Close()

}

// 错误处理函数

func (r *RabbitMQ) failOnErr(err error, message string) {

if err != nil {

log.Fatalf(“%s:%s”, message, err)

panic(fmt.Sprintf(“%s:%s”, message, err))

}

// NewRabbitMQRouting 路由模式创建RabbitMQ实例

func NewRabbitMQRouting(exchangeName string, routingKey string) *RabbitMQ {

//创建RabbitMQ实例

rabbitmq := NewRabbitMQ(“”, exchangeName, routingKey)

var err error

//获取connection

rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)

rabbitmq.failOnErr(err, “failed to connect rabbitmq!”)

//获取channel

rabbitmq.channel, err = rabbitmq.conn.Channel()

rabbitmq.failOnErr(err, “failed to open a channel”)

return rabbitmq

}

// PublishRouting 路由模式发送消息

func (r *RabbitMQ) PublishRouting(message string) {

//1.尝试创建交换机

err := r.channel.ExchangeDeclare(

//交换机名称

r.Exchange,

//交换机类型

“direct”,

//是否持久化

true,

//是否自动删除

false,

//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定

false,

//是否阻塞处理

false,

//其他属性

nil,

)

r.failOnErr(err, “Failed to declare an exchange”)

//2.发送消息
//发送消息
//参数：
//1.交换机
//2.队列，
//3.是否强制性路由到队列，如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还
//给发送者
//4.是否立即发送
//5.其他属性
//6.发送消息的内容
//调用channel 发送消息到队列中
r.channel.Publish(
    r.Exchange,
    r.Key,
    //如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
    false,
    //如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
    false,
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(message),
    })

}

// ConsumeRouting routing 模式下消费者

func (r *RabbitMQ) ConsumeRouting() {

//1.试探性创建交换机

err := r.channel.ExchangeDeclare(

// 交换机名称

r.Exchange,

//交换机类型

“direct”,

//是否持久化

true,

// 是否自动删除

false,

//YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定

false,

nil,

)

r.failOnErr(err, “Failed to declare an exchange”)

//2.试探性创建队列，这里注意队列名称不要写
q, err := r.channel.QueueDeclare(
    //随机生产队列名称
    "",
    //是否持久化
    false,
    //是否自动删除
    false,
    //是否具有排他性
    true,
    //是否阻塞处理
    false,
    //其他参数
    nil,
)
r.failOnErr(err, "Failed to declare a queue")
//绑定队列到 exchange 中
err = r.channel.QueueBind(
    q.Name,
    // routing模式下要指定key
    r.Key,
    r.Exchange,
    false,
    nil)
//消费消息
messages, err := r.channel.Consume(
    q.Name,
    "",
    true,
    false,
    false,
    false,
    nil,
)
forever := make(chan bool)
//启用协程处理消息
go func() {
    for d := range messages {
        log.Printf("Received a message: %s", d.Body)
    }
}()
fmt.Println("退出请按 CTRL+C\n")
<-forever

}

##### 4.2.2. mainRoutingPublish.go

package main

import (

“fmt”

“qiutian.com/study/02.rabbitmq/Routing/mq”

“strconv”

“time”

)

func main() {

routingOne := mq.NewRabbitMQRouting("newRoutingExchange", "newRoutingKey_one")
routingTwo := mq.NewRabbitMQRouting("newRoutingExchange", "newRoutingKey_two")
for i := 0; i < 100; i++ {
    if i%2 == 0 {
        routingOne.PublishRouting("Hello World! " + strconv.Itoa(i))
    } else {
        routingTwo.PublishRouting("Hello World! " + strconv.Itoa(i))
    }
    time.Sleep(time.Second)
    fmt.Println(i)
}

}

##### 4.2.3. mianRoutingRecieve1.go

package main

import (

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加戳这里获取

package main

import (

“fmt”

“qiutian.com/study/02.rabbitmq/Routing/mq”

“strconv”

“time”

)

func main() {

routingOne := mq.NewRabbitMQRouting("newRoutingExchange", "newRoutingKey_one")
routingTwo := mq.NewRabbitMQRouting("newRoutingExchange", "newRoutingKey_two")
for i := 0; i < 100; i++ {
    if i%2 == 0 {
        routingOne.PublishRouting("Hello World! " + strconv.Itoa(i))
    } else {
        routingTwo.PublishRouting("Hello World! " + strconv.Itoa(i))
    }
    time.Sleep(time.Second)
    fmt.Println(i)
}

}

##### 4.2.3. mianRoutingRecieve1.go

package main

import (

[外链图片转存中…(img-BIeCOMzy-1715401866579)]

[外链图片转存中…(img-MfaIoGye-1715401866579)]

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加戳这里获取

分类:热门推荐日期:2024-06-25浏览:1评论:0

队列 消息

相关推荐