RabbitMQ使用随笔

1.RabbitMQ介绍

RabbitMQ 是一个开源的消息代理软件，采用了先进的消息队列协议 AMQP（Advanced Message Queuing Protocol）。它主要用于处理分布式系统中各个组件间的消息传递和通信。RabbitMQ 支持多种协议，但 AMQP 是其最常用和核心的协议。

主要特性：

高可用性：RabbitMQ 支持集群部署，能保证消息服务的高可用性，甚至在某些节点发生故障时，系统也能继续提供服务。
消息持久化：它支持将消息持久化到磁盘，以确保在系统崩溃时不会丢失消息。
可靠传递：RabbitMQ 确保消息可以可靠地传送到消费者，使用确认机制避免消息丢失。
消息路由：通过交换机（Exchange）来路由消息，可以根据不同的规则将消息传送到不同的队列。
多协议支持：除了 AMQP，RabbitMQ 还支持其他协议，如 MQTT、STOMP 和 HTTP 等。
灵活的消息队列模型：支持多种队列类型，如直接队列（Direct）、扇形队列（Fanout）、主题队列（Topic）和头部队列（Header）。

架构特点：

生产者（Producer）：负责发送消息。
消费者（Consumer）：负责接收和处理消息。
队列（Queue）：存储消息的地方，消费者从队列中获取消息。
交换机（Exchange）：决定如何将消息路由到队列。
绑定（Binding）：交换机与队列之间的连接。

2.应用场景：

异步任务处理：在分布式系统中，通过 RabbitMQ 将任务发送到不同的处理单元进行处理，从而解耦各个服务。
消息广播：通过 RabbitMQ 的扇形交换机，将消息广播给多个消费者。
流量削峰：通过将消息放入队列并按顺序处理，可以有效削减瞬时流量带来的压力。
事件驱动架构：使用 RabbitMQ 实现不同模块间的事件通知和传递。

3.消息队列模型

3.1 简单模式

普通队列模型，生产者消费者直接绑定到队列上(默认交换机)，一个生产，一个消费。

队列模型图:

普通一对一示列代码:

producter:

package mainimport ("fmt""log""time""github.com/streadway/amqp"
)func main() {// 连接到 RabbitMQconn, err := amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")if err != nil {log.Fatalf("failed to connect to RabbitMQ: %v", err)}defer conn.Close()// 创建一个通道ch, err := conn.Channel()if err != nil {log.Fatalf("failed to open a channel: %v", err)}defer ch.Close()// 定义一个队列q, err := ch.QueueDeclare("task_queue", // 队列名称true,         // 是否持久化false,        // 是否自动删除false,        // 是否排他性false,        // 是否等待nil,          // 额外属性)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 发送消息i := 0for {i += 1body := fmt.Sprintf("Welcome mq world ! #%d", i)err = ch.Publish("",     // 默认交换机q.Name, // 队列名称true,   // 是否持久化false,  // 是否立即发送amqp.Publishing{ContentType:  "text/plain",Body:         []byte(body),DeliveryMode: amqp.Persistent, // 消息持久化})if err != nil {log.Fatalf("failed to publish a message: %v", err)}fmt.Printf("Sent: %s\n", body)time.Sleep(1 * time.Second) // 每秒发送一个消息}// 关闭连接fmt.Println("All messages sent.")
}

comsumer:

package mainimport ("fmt""log""github.com/streadway/amqp"
)func main() {// 连接到 RabbitMQconn, err := amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")if err != nil {log.Fatalf("failed to connect to RabbitMQ: %v", err)}defer conn.Close()// 创建一个通道ch, err := conn.Channel()if err != nil {log.Fatalf("failed to open a channel: %v", err)}defer ch.Close()// 定义一个队列q, err := ch.QueueDeclare("task_queue", // 队列名称true,         // 是否持久化false,        // 是否自动删除false,        // 是否排他性false,        // 是否等待nil,          // 额外属性)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 创建一个消费者msgs, err := ch.Consume(q.Name,     // 队列名称"EricTest", // 消费者名称//true,       // 是否自动确认,只要消费这节收到，就标记其可以删除(适用于消费者能高效的处理消息场景)false, // 采用手动确认,自定义消息应答模式，做到收到消息且消息真正处理完了，再告知消息可以删除了。(还支持批量应答)false, // 是否排他性false, // 是否阻塞false, // 是否等待nil,   // 额外属性)if err != nil {log.Fatalf("failed to register a consumer: %v", err)}// 设置消息处理策略err = ch.Qos(1, 0, false)if err != nil {log.Fatalf("failed to set QoS: %v", err)}// 使用 goroutine 处理消息go func() {for msg := range msgs {fmt.Printf("Received a message: %s\n", msg.Body)// 处理消息// 手动应答// 手动确认,设置为false,表示不重新排队,如果处理失败情况,可以通过msg.Nack来拒绝并重新入队/*msg.Nack消息处理失败,重新入队if err := msg.Nack(false,true);err != nil {}*/if err := msg.Ack(false); err != nil {log.Fatalf("failed to acknowledge message: %v", err)} else {fmt.Printf("Acknowledged message: %s\n", msg.Body)}}}()// 阻塞主线程，保持服务端一直运行fmt.Println("Waiting for messages. To exit press CTRL+C")select {}
}

还可以通过一个P和多个C，多个消费者从同一个队列中获取消息进行消费，默认情况是平均分配的消费，不会因为那个消费块就让其多消费，模拟两个消费，观察现象（C1消费快，C2消费慢），如上代码，启动一个P和两个C模拟观察情况,模拟输出如下:

同一份消费代码：

C1比C2慢速率慢2秒：(上述代码消费端加个延时即可)

3.2 队列绑定交换机模式(以及几种交换机特点)

图解模型：

该模型的特点是，生产者的消息不再直接发往队列，而是发送到交换机，再由交换机根据规则将消息传送到队列汇总。这里解释下交换机概念?

队列交换机:

定义的一个消息组建，其目的是负责接收生产者发送的消息，并根据一定的规则将消息路由到一个或者多个队列中。模型如上图所示，他主要是根据路由键和绑定规则来确定消息的去向。在rabbitmq中，提供了Direct(直接交换机)，Fanout(广播交换机)，Topic(主题交换机)，Headers(头交换机)四种类型，每种类型机制不一，具体如下：

direct：

a.消息会被路由到与指定路由键完全匹配的队列。

b.如果交换机的路由键与队列的绑定路由键匹配，消息将被发送到该队列

定向direct，一个交换机，两个队列，不同的路由key匹配,模拟输出,两个队列匹配不同的路由key，交换机每次根据具体的路由key将消息推送到不同的队列中

fanout:

a.消息会被路由到所有与该交换机绑定的队列，忽略虑路由键

b.这种类型的交换机类似广播，所有绑定的队列都会接收到消息

广播fanout，通过交换节的消息向所有绑定的队列都推送一份消息，模拟输出：

topic:

a.路由键是一个由多个部分组成的字符串.（person.rabbit.white）

b.交换机会根据绑定时使用的模式（通配符 * 和 #）来选择匹配的队列。

c.* 用来匹配单个词，# 用来匹配零个或多个词

Topic通过模式匹配转发消息且消费：

headers:

a.消息根据其头部信息（而非路由键）来进行路由

b.交换机会根据队列与交换机之间的绑定条件（头部的键值对）来匹配消息

headers模式模拟代码:

代码部分片段:p&c

producter:

// 创建一个通道ch, err := conn.Channel()if err != nil {log.Fatalf("failed to open a channel: %v", err)}defer ch.Close()// 声明一个交换机err = ch.ExchangeDeclare("headers_exchange", // 交换机名称"headers",          // 交换机类型,主题方式false,              // 是否持久化false,              // 是否自动删除false,              // 是否内部使用false,              // 是否等待确认nil,                // 额外参数)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 发布消息到交换机,指定路由键//routerKey := "info_eric"// 发送消息i := 0for {i += 1body := fmt.Sprintf("Welcome mq world ! #%d", i)err = ch.Publish("headers_exchange", //交换机名称//q.Name,          // 队列名称"",    // 路由键 headers模式下不使用false, // 是否持久化false, // 是否立即发送amqp.Publishing{ContentType:  "text/plain",Body:         []byte(body),DeliveryMode: amqp.Transient, // 消息持久化Headers: amqp.Table{"X-Message-Type": "" +"greeting", // 设置消息头},})if err != nil {log.Fatalf("failed to publish a message: %v", err)}fmt.Printf("Sent: %s\n", body)time.Sleep(1 * time.Second) // 每秒发送一个消息}

comsumer:

// 创建一个通道ch, err := conn.Channel()if err != nil {log.Fatalf("failed to open a channel: %v", err)}defer ch.Close()// 声明一个队列q, err := ch.QueueDeclare("headers_queue", // 队列名称false,           // 是否持久化false,           // 是否自动删除false,           // 是否排他性false,           // 是否等待nil,             // 额外属性)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 声明一个交换机err = ch.ExchangeDeclare("headers_exchange", // 交换机名称"headers",          // 交换机类型,广播false,              // 是否持久化false,              // 是否自动删除false,              // 是否内部使用false,              // 是否等待确认nil,                // 额外参数)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 绑定队列到交换机err = ch.QueueBind(q.Name,             // 队列名称"",                 // 路由键,头部消息路由,该字段不使用"headers_exchange", // 交换机名称false,              // 是否等待确认nil,                // 额外参数)// 创建一个消费者msgs, err := ch.Consume(q.Name,          // 队列名称"comsumer_eric", // 消费者名称//true,       // 是否自动确认,只要消费这节收到，就标记其可以删除(适用于消费者能高效的处理消息场景)false, // 采用手动确认,自定义消息应答模式，做到收到消息且消息真正处理完了，再告知消息可以删除了。(还支持批量应答)false, // 是否排他性false, // 是否阻塞false, // 是否等待nil,   // 额外属性)if err != nil {log.Fatalf("failed to register a consumer: %v", err)}// 设置消息处理策略err = ch.Qos(1, 0, false)if err != nil {log.Fatalf("failed to set QoS: %v", err)}// 使用 goroutine 处理消息go func() {for msg := range msgs {fmt.Printf("Received a message: %s %s\n", msg.Body, msg.Headers)// 处理消息// 手动应答// 手动确认,设置为false,表示不重新排队,如果处理失败情况,可以通过msg.Nack来拒绝并重新入队/*msg.Nack消息处理失败,重新入队if err := msg.Nack(false,true);err != nil {}*/if err := msg.Ack(false); err != nil {log.Fatalf("failed to acknowledge message: %v", err)} else {fmt.Printf("Acknowledged message: %s\n", msg.Body)}}}()

模拟输出：

综上，交换机的功能将消息路由到队列，且自身不存储消息。另外，不同的类型的交换根据不同的路由策略将消息分发到合适的队列中去。

以上几种情况示例代码如下可以灵活修改模拟实现上述规则，另外还列举出了定向direct方式下只想某种匹配的路由key下发消息情况，请见如下示例：

producter:

package mainimport ("fmt""log""time""github.com/streadway/amqp"
)func main() {// 连接到 RabbitMQconn, err := amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")if err != nil {log.Fatalf("failed to connect to RabbitMQ: %v", err)}defer conn.Close()// 创建一个通道ch, err := conn.Channel()if err != nil {log.Fatalf("failed to open a channel: %v", err)}defer ch.Close()// 定义一个交换机err = ch.ExchangeDeclare("exchange_eric", // 交换机名称"direct",        // 交换机类型true,            // 是否持久化false,           // 是否自动删除false,           // 是否内部使用false,           // 是否等待确认nil,             // 额外参数)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 发布消息到交换机,指定路由键routerKey := "info_eric"// 发送消息i := 0for {i += 1body := fmt.Sprintf("Welcome mq world ! #%d", i)err = ch.Publish("exchange_eric", //交换机名称routerKey, // 指定路由键true,      // 是否持久化false,     // 是否立即发送amqp.Publishing{ContentType:  "text/plain",Body:         []byte(body),DeliveryMode: amqp.Persistent, // 消息持久化})if err != nil {log.Fatalf("failed to publish a message: %v", err)}fmt.Printf("Sent: %s\n", body)time.Sleep(1 * time.Second) // 每秒发送一个消息}// 关闭连接fmt.Println("All messages sent.")
}

comsumer: (当前消费者绑定与生产者匹配的路由键info_eric)

package mainimport ("fmt""log""github.com/streadway/amqp"
)func main() {// 连接到 RabbitMQconn, err := amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")if err != nil {log.Fatalf("failed to connect to RabbitMQ: %v", err)}defer conn.Close()// 创建一个通道ch, err := conn.Channel()if err != nil {log.Fatalf("failed to open a channel: %v", err)}defer ch.Close()// 定义一个队列q, err := ch.QueueDeclare("task_queue_zero", // 队列名称true,         // 是否持久化false,        // 是否自动删除false,        // 是否排他性false,        // 是否等待nil,          // 额外属性)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 定义一个交换机err = ch.ExchangeDeclare("exchange_eric", // 交换机名称"direct",        // 交换机类型true,            // 是否持久化false,           // 是否自动删除false,           // 是否内部使用false,           // 是否等待确认nil,             // 额外参数)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 绑定队列到交换机err = ch.QueueBind(q.Name,          // 队列名称"info_eric",     // 路由键,绑定是制定具体路由键,生产者发送消息时，指定路由键,将消息按照对应路由键下发到队列当中"exchange_eric", // 交换机名称false,           // 是否等待确认nil,             // 额外参数)// 创建一个消费者msgs, err := ch.Consume(q.Name,          // 队列名称"comsumer_eric", // 消费者名称//true,       // 是否自动确认,只要消费这节收到，就标记其可以删除(适用于消费者能高效的处理消息场景)false, // 采用手动确认,自定义消息应答模式，做到收到消息且消息真正处理完了，再告知消息可以删除了。(还支持批量应答)false, // 是否排他性false, // 是否阻塞false, // 是否等待nil,   // 额外属性)if err != nil {log.Fatalf("failed to register a consumer: %v", err)}// 设置消息处理策略err = ch.Qos(1, 0, false)if err != nil {log.Fatalf("failed to set QoS: %v", err)}// 使用 goroutine 处理消息go func() {for msg := range msgs {fmt.Printf("Received a message: %s\n", msg.Body)// 处理消息// 手动应答// 手动确认,设置为false,表示不重新排队,如果处理失败情况,可以通过msg.Nack来拒绝并重新入队/*msg.Nack消息处理失败,重新入队if err := msg.Nack(false,true);err != nil {}*/if err := msg.Ack(false); err != nil {log.Fatalf("failed to acknowledge message: %v", err)} else {fmt.Printf("Acknowledged message: %s\n", msg.Body)}}}()// 阻塞主线程，保持服务端一直运行fmt.Println("Waiting for messages. To exit press CTRL+C")select {}
}

另外队列设定其他路由键，虽然是同一个交换机，但是交换机派发消息按照路由键匹配规则下发:

	// 定义一个队列q, err := ch.QueueDeclare("task_queue_one", // 队列名称true,         // 是否持久化false,        // 是否自动删除false,        // 是否排他性false,        // 是否等待nil,          // 额外属性)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 定义一个交换机err = ch.ExchangeDeclare("exchange_eric", // 交换机名称"direct",        // 交换机类型true,            // 是否持久化false,           // 是否自动删除false,           // 是否内部使用false,           // 是否等待确认nil,             // 额外参数)if err != nil {log.Fatalf("failed to declare a queue: %v", err)}// 绑定队列到交换机err = ch.QueueBind(q.Name,          // 队列名称"info_jean",     // 路由键"exchange_eric", // 交换机名称false,           // 是否等待确认nil,             // 额外参数)

上述代码中路由键跟consumer不一致，所以当前的队列task_queue_one不会接收到producter推送的消息,模拟输出如下:

上图所示,路由键匹配的队列能接受到消息，不匹配的因无消息消费阻塞等待。

producter代码优化，设置两种路由key随机的分发消息到交换机，交换机根绝每个队列匹配的路由key将消息分发到不同的队列中，实际业务中可利用不同的路由key将消息转发。

修改后的producter代码，注意routerKeys的相关部分：

// 发布消息到交换机,指定路由键routerKeys := []string{"info_jean", "info_eric"}// 发送消息i := 0for {i += 1body := fmt.Sprintf("Welcome mq world ! #%d", i)routerKeyIndex := rand.Intn(2)err = ch.Publish("exchange_eric", //交换机名称routerKeys[routerKeyIndex], // 指定路由键true,      // 是否持久化false,     // 是否立即发送amqp.Publishing{ContentType:  "text/plain",Body:         []byte(body),DeliveryMode: amqp.Persistent, // 消息持久化})if err != nil {log.Fatalf("failed to publish a message: %v", err)}fmt.Printf("Sent: %s\n", body)time.Sleep(1 * time.Second) // 每秒发送一个消息}

模拟实现输出: