ToB企服应用市场:ToB评测及商务社交产业平台

标题: Golang学习条记_RabbitMQ的原理架构和使用 [打印本页]

---

作者: 小秦哥    时间: 2024-6-19 20:40
标题: Golang学习条记_RabbitMQ的原理架构和使用
RabbitMQ 简介

• 实现了高级消息队列协议（Advanced Message Queuing Protcol）AMQP
  
• 消息队列中央件的作用（Redis实现MQ里面有写过，这里简朴带过）
  
  
  • 解耦
    
  • 削峰
    
  • 异步处理
    
  • 缓存
    
  • 消息通信
    
  • 提高扩展性
    
  
  

RabbitMQ 架构明确

     

• binding(绑定)：互换机将消息路由给Queue所遵循的规则，可以界说一个路由键，用于互换机筛选特定的Queue
  
  
  • Routing_Key(路由键)：Producer 和 Consumer 协商一致的 key 策略。主要在互换机的 direct(直连)和 topic(主题) 模式下使用，fanout(广播)模式下不使用Routing_Key
    
  
  
• Exchange(互换机)：主要功能是分发消息给特定的Queue，只负责转发，不具备存储消息的功能。Exchange有以下四种模式：
  
  
  • direct(直连模式)，根据携带的Routing_Key来筛选特定的Queue进行消息投递。是RabbitMQ的默认类型，可以不指定Routing_Key，在创建时会默认生成与Queue重名。
    
  • hander(头模式)，使用场景不多，消息路由涉及多个属性的时候，互换机使用多属性来取代Routing_key创建路由规则，还可以界说匹配单词的个数，例如any为有一个单词满意条件就匹配乐成。all为所有单词都满意条件才匹配乐成。
    
  • fanout(广播模式)，不看Routing_Key。只根据Exchange和Queue的binding情况来分发信息。所有与之binding的queue都将接收到同一条消息。
    
  • topic(主题模式)，相当于模糊查询。topic的routing_key是使用 . 来进行隔断的。有两种匹配方法：
    
    
    • " * " 匹配一个单词，例子如下
      
    • " # " 匹配0个~多个单词，例子如下
      
    
    
  
  

1. rabbitMQ.* == rabbitMQ.topic != rabbitMQ.topic.topic
2. rabbitMQ.# == rabbit.topic == rabbit.topic.topic

复制代码

• Queue（消息队列的存储数据布局）：
  
  
  • 存储方式：
    
    
    
    • 持久化，在Server当地硬盘存储一份
      
    • 临时队列，重启后丢失数据
      
    • 自动删除，不存在用户连接则删除queue
      
    
    
  • 队列对ACK请求的不同情况
    
    
    
    • consumer 接收并 ack，queue 删除数据并向 consumer 发送新消息
      
    • consumer 接收但是未 ack 就断开了连接，queue 会认为消息并未传送乐成，consumer 再次连接时会重新发送消息
      
    • 如果consumer 接收消息乐成 ，但是忘记 ack 则 queue 不会重复发送消息
      
    • 如果 consumer 拒收消息，则 queue 会向另外满意条件的 consumer 继承发送这条消息
      
    
    
  
  

RabbitMQ 工作流程

Producer方向

• Producer 与 RabbitMQ Broker 创建连接，开启一个信道 channel
  
• 声明互换机并设置属性（互换机类型、持久化等）
  
• 声明Queue并设置属性（持久化，自动删除等）
  
• 通过Routing_key来binding互换机和Queue
  
• 发送信息给互换，互换机根据Routing_key来确认投递的queue
  
• 查找乐成后将消息存到queue
  
• 查找失败将消息丢弃或抛回给生产者
  
• 关闭channel
  

Consumer方向

• 与 queue 创建连接，开启channel
  
• 向queue请求队列中的msg
  
• 等待queue回应，开始接收消息
  
• 消息处理完成后 返回回调确认ack
  
• queue 将确认的消息从队列中删除
  
• 关闭channel
  

RabbitMQ的两种摆设方式

Meta Data ： 元数据（形貌数据的数据）

• vhost meta data : 为Queue、Exchange、Binding提供命名空间级别的隔离
  
• exchange meta data：记录路由的名称类型和属性
  
• binding mate data：映射 routing_key和queue之间的绑定关系
  
• queue mate data：表队列名称和属性
  

普通模式

对于该模式的两个节点，消息只会存在其中一个节点，另一个节点只保存mate data，当consumer 连接节点2访问节点1的数据信息时，消息会在两个节点中传递。
该模式下p和c应尽量连接每个节点，这样起到线性拓展的作用。
但存在一个问题，如果节点上另有未消费的消息，但是节点挂了。如果节点设置了持久化，则需要在节点重启的时候消息才会规复。如果未设置持久化，则消息会丢失。
镜像模式

消息存在多个节点中，消息会在节点与节点之间同步，可实现高可用（当一个节点挂了，另一个节点可以接替其位置，继承工作）但会低落性能，由于大量消息进入和同步，会占用大量带宽，但是为了保证高可靠性需要弃取。
面试题

• Q：如何保证消息不被重复消费？
  
  
  
  • A：MQ通过确认机制ACK，进行确认。确认后消息从queue中删除，保证消息不被重复消费的。如果由于网络原因ack没有乐成发出，导致消息重新投递。可以使用全局唯一消息id来制止。
    
     
  
  
  • 消息发送者发送消息时携带一个全局唯一的消息id
    
  • 消费者监听到消息后，根据id在redis或者db中查询是否存在消费记录
    
  • 如果没有消费就正常消费，消费完毕后，写入redis或者db
    
  • 如果消息消费过则直接丢弃
    
  
  
• Q：如何保证消息的消费次序？
  
  
  
  • A：RabbitMQ中存在一个设置，叫独占队列。即在同一时间只有一个消费者会消费消息。从而制止了异步操作，保证消费次序。或者一个Producer对一个Consumer
    
  
  
• Q：如何保证数据一致性？
  
  
  
  • A：由于MQ的使用场景多为分布式系统，以是一般不寻求强一致性。而保证终极一致性就可以。
    
  • 而保证数据终极一致性，可以采用消息赔偿机制。即消息在消费者处理完之后调用生产者的API修改数据状态。如未调用API则判断为消息处理失败或出错。此时间隔一段时间后重新投递消息进行再次操作。
    
  • 消费者收到消息，处理完毕后，发送一条响应消息给生产者也是消息赔偿机制，本意是确认消费者乐成消费消息。ACK也是处理方法
    
  
  

RabbitMQ的使用（Golang使用amqp包）

代码部分参考 upup小亮的博客
代码只是简朴的操作，主要是认识流程。对于如何创建Queue和绑定Exchange之类的操作有个相识。

Simple（简朴收发模式，只有一个Queue）

Simple运行机制与WorkQueue相似，只是一个Consumer与多个Consumer的区别。多个Consumer之间存在竞争关系，以是工作队列是创建多个Consumer，多个竞争只有一个可以获取消息消费。消费乐成后ack消息删除。
演示代码放到一起了：
WorkQueue 工作队列

生产者

1. // simple and work queue
2. func main2() {
3.         // 连接到 rabbitMQ
4.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
5.         if err != nil {
6.                 log.Fatalf("无法创建连接:%s", err)
7.                 return
8.         }
9.         // 默认关闭
10.         defer conn.Close()
12.         // 创建通道Channel
13.         ch, err := conn.Channel()
14.         if err != nil {
15.                 log.Fatalf("无法创建channel:%s", err)
16.                 return
17.         }
18.         // 通道关闭
19.         defer ch.Close()
21.         // 创建存储队列
22.         queue, err := ch.QueueDeclare(
23.                 "hello", // 队列名称
24.                 false, // 持久化设置，可以为true根据需求选择
25.                 false, // 自动删除，没有用户连接删除queue一般不选用
26.                 false, //独占
27.                 false, //等待服务器确认
28.                 nil)   //参数
29.         if err != nil {
30.                 fmt.Println(err)
31.                 log.Fatalf("无法声明队列:%s", err)
32.                 return
33.         }
35.         var body string
36.         // 发送信息
37.         for i := 0; i < 10; i++ {
38.                 fmt.Println(i)
39.                 body = "Hello RabbitMQ" + string(i)
40.                 err = ch.Publish(
41.                         "",
42.                         queue.Name,
43.                         false, // 必须发送到消息队列
44.                         false, // 不等待服务器确认
45.                         amqp.Publishing{
46.                                 ContentType: "text/plain",
47.                                 Body:        []byte(body),
48.                         })
49.                 if err != nil {
50.                         log.Fatalf("消息生产失败:%s", err)
51.                         continue
52.                 }
53.         }
54. }

复制代码

消费者

1.         // create conn
2.         // 如果同时运行两个这样的consumer代码，就是工作队列。只有一个consumer就是simple
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil {
5.                 log.Fatalf("无法创建连接:%s", err)
6.                 return
7.         }
8.         defer conn.Close()
10.         // create channel
11.         ch, err := conn.Channel()
12.         if err != nil {
13.                 log.Fatalf("无法创建channel:%s", err)
14.                 return
15.         }
16.         defer ch.Close()
17.         // create queue
18.         queue, err := ch.QueueDeclare(
19.                 "hello",
20.                 false,
21.                 false,
22.                 false,
23.                 false,
24.                 nil)
25.         if err != nil {
26.                 log.Fatalf("无法创建queue:%s", err)
27.                 return
28.         }
29.        
30.         // 消费信息
32.         msgs, err := ch.Consume(
33.                 queue.Name,
34.                 "",
35.                 true,
36.                 false,
37.                 false,
38.                 false,
39.                 nil)
40.         if err != nil {
41.                 log.Fatalf("无法消费信息:%s", err)
42.                 return
43.         }
44.         for msg := range msgs {
45.                 log.Println(string(msg.Body))
46.         }
47.         return

复制代码

pub/sub 发布订阅模式

发布订阅模式可以创建两个Queue，绑定到同一个Exchange中
生产者这边只需要跟互换机对接，而互换机类型为fanout：

1. func main() {
2.         // 连接到 rabbitMQ
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil {
5.                 log.Fatalf("无法创建连接:%s", err)
6.         }
7.         // 默认关闭
8.         defer conn.Close()
10.         // 创建通道Channel
11.         ch, err := conn.Channel()
12.         if err != nil {
13.                 log.Fatalf("无法创建channel:%s", err)
14.         }
15.         defer ch.Close()
17.         // create exchange
18.         ex := ch.ExchangeDeclare(
19.                 "exchange1", // 交换机名称
20.                 "fanout",    // 交换机类型
21.                 true,        // 是否持久化
22.                 false,       // 是否自动删除
23.                 false,       // 是否内部使用
24.                 false,       // 是否等待服务器响应
25.                 nil,         // 其他属性
26.         )
27.         fmt.Println(ex)
29.         body := "Hello RabbitMQ for Pub/Sub"
30.         err = ch.Publish(
31.                 "exchange1",
32.                 "", // routing key 可以为空，因为fanout不看routing key
33.                 false,
34.                 false,
35.                 amqp.Publishing{
36.                         ContentType: "text/plain",
37.                         Body:        []byte(body),
38.                 })
39.         if err != nil {
40.                 log.Fatalf("err %s:", err)
41.         }
42.         log.Println(body)
43. }

复制代码

消费者：创建互换机，类型为fanout，创建队列，绑定互换机（创建多个consumer绑定同一个queue和同一个互换机。这样发送一个消息，所有的consumer都能收到。== 发布订阅模型）

1.         // Pub/Sub
2.         // Create conn
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil{
5.                 log.Fatalf(err)
6.         }
7.         defer conn.Close()
9.         // channel create
10.         ch, err := conn.Channel()
11.         if err != nil{
12.                 log.Fatalf(err)
13.         }
14.         defer ch.Close()
16.         // exchange create
17.         ex := ch.ExchangeDeclare(
18.                 "exchange1",
19.                 "fanout",
20.                 true,
21.                 false,
22.                 false,
23.                 false,
24.                 nil)
26.         fmt.Println(ex)
28.         // queue create
29.         queue, err := ch.QueueDeclare(
30.                 "hello",
31.                 false,
32.                 false,
33.                 false,
34.                 false,
35.                 nil)
36.         if err != nil{
37.                 log.Fatalf(err)
38.         }
39.         err = ch.QueueBind(
40.                 queue.Name,
41.                 "",
42.                 "exchange1",
43.                 false,
44.                 nil)
45.         if err != nil{
46.                 log.Fatalf(err)
47.         }
49.         msgs, err := ch.Consume(
50.                 queue.Name,
51.                 "",
52.                 true,
53.                 false,
54.                 false,
55.                 false,
56.                 nil)
57.         if err != nil{
58.                 log.Fatalf(err)
59.         }
60.         go func() {
61.                 for d := range msgs {
62.                         log.Printf("Received a message: %s", d.Body)
63.                 }
64.         }()
66.         log.Printf("Waiting for messages. To exit press CTRL+C")
67.         <-make(chan struct{}) // 阻塞主goroutine
68. }

复制代码

Routing 模式（对特定的队列投递消息）

生产者

1. func main() {
2.         // 连接到 rabbitMQ
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil {
5.                 log.Fatalf("无法创建连接:%s", err)
6.         }
7.         // 默认关闭
8.         defer conn.Close()
10.         // 创建通道Channel
11.         ch, err := conn.Channel()
12.         if err != nil {
13.                 log.Fatalf("无法创建channel:%s", err)
14.         }
15.         defer ch.Close()
17.         // create exchange
18.         ex := ch.ExchangeDeclare(
19.                 "exchange1", // 交换机名称
20.                 "direct",    // 交换机类型
21.                 true,        // 是否持久化
22.                 false,       // 是否自动删除
23.                 false,       // 是否内部使用
24.                 false,       // 是否等待服务器响应
25.                 nil,         // 其他属性
26.         )
27.         fmt.Println(ex)
28.         body := "Hello RabbitMQ for direct routing"
29.                 // 发布消息到交换机，并指定路由键
30.         err = ch.Publish(
31.                 "logs_direct", // 交换机名称
32.                 "routing_key", // 路由键
33.                 false,         // 是否等待服务器响应
34.                 false,         // 是否立即将消息写入磁盘
35.                 amqp.Publishing{
36.                         ContentType: "text/plain",
37.                         Body:        []byte(body),
38.                 },
39.         )
40.         if err != nil{
41.                 log.Fatalf("无法创建send msg:%s", err)
42.         }
43.         log.Printf("Sent message: %s", message)

复制代码

消费者

1. func main() {
2.         // 连接到RabbitMQ服务器
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil{
5.                 log.Fatalf("无法创建send msg:%s", err)
6.         }
7.         defer conn.Close()
9.         // 创建一个通道
10.         ch, err := conn.Channel()
11.         if err != nil{
12.                 log.Fatalf("无法创建send msg:%s", err)
13.         }
14.         defer ch.Close()
16.         // 声明一个交换机
17.         err = ch.ExchangeDeclare(
18.                 "logs_direct", // 交换机名称
19.                 "direct",      // 交换机类型
20.                 true,          // 是否持久化
21.                 false,         // 是否自动删除
22.                 false,         // 是否内部使用
23.                 false,         // 是否等待服务器响应
24.                 nil,           // 其他属性
25.         )
26.         if err != nil{
27.                 log.Fatalf("无法创建send msg:%s", err)
28.         }
30.         // 声明一个临时队列
31.         q, err := ch.QueueDeclare(
32.                 "",    // 队列名称，留空表示由RabbitMQ自动生成,因为定义了key所以队列名可以是随意的，毕竟是依靠key来进行匹配的
33.                 false, // 是否持久化
34.                 false, // 是否自动删除（当没有任何消费者连接时）
35.                 true,  // 是否排他队列（仅限于当前连接）
36.                 false, // 是否等待服务器响应
37.                 nil,   // 其他属性
38.         )
39.         // 将队列绑定到交换机上，并指定要接收的路由键
40.         err = ch.QueueBind(
41.                 q.Name,        // 队列名称
42.                 "routing_key",      // 路由键
43.                 "logs_direct", // 交换机名称
44.                 false,         // 是否等待服务器响应
45.                 nil,           // 其他属性
46.         )
47.         if err != nil{
48.                 log.Fatalf("无法创建send msg:%s", err)
49.         }
51.         // 订阅消息
52.         msgs, err := ch.Consume(
53.                 q.Name, // 队列名称
54.                 "",     // 消费者标识符，留空表示由RabbitMQ自动生成
55.                 true,   // 是否自动应答
56.                 false,  // 是否独占模式（仅限于当前连接）
57.                 false,  // 是否等待服务器响应
58.                 false,  // 其他属性
59.                 nil,    // 其他属性
60.         )
61.         failOnError(err, "Failed to register a consumer")
63.         // 接收消息的goroutine
64.         go func() {
65.                 for d := range msgs {
66.                         log.Printf("Received a message: %s", d.Body)
67.                 }
68.         }()
70.         log.Printf("Waiting for messages. To exit press CTRL+C")
71.         <-make(chan struct{}) // 阻塞主goroutine

复制代码

topic

1. func main() {
2.         // 连接到RabbitMQ服务器
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil{
5.                 log.Fatalf(err)
6.         }
7.         defer conn.Close()
9.         // 创建一个通道
10.         ch, err := conn.Channel()
11.         if err != nil{
12.                 log.Fatalf(err)
13.         }
14.         defer ch.Close()
16.         // 声明一个交换机
17.         err = ch.ExchangeDeclare(
18.                 "logs_topic", // 交换机名称
19.                 "topic",      // 交换机类型
20.                 true,         // 是否持久化
21.                 false,        // 是否自动删除
22.                 false,        // 是否内部使用
23.                 false,        // 是否等待服务器响应
24.                 nil,          // 其他属性
25.         )
26.         if err != nil{
27.                 log.Fatalf(err)
28.         }
30.         // 定义要发送的消息的路由键和内容
31.         routingKey := "example.key.das"
32.         message := "Hello, RabbitMQ!"
34.         // 发布消息到交换机，并指定路由键
35.         err = ch.Publish(
36.                 "logs_topic", // 交换机名称
37.                 routingKey,   // 路由键
38.                 false,        // 是否等待服务器响应
39.                 false,        // 是否立即发送
40.                 amqp.Publishing{
41.                         ContentType: "text/plain",
42.                         Body:        []byte(message),
43.                 },
44.         )
45.         if err != nil{
46.                 log.Fatalf(err)
47.         }
49.         log.Printf("Sent message: %s", message)
50. }

复制代码

消费者

1. func main() {
2.         // 连接到RabbitMQ服务器
3.         conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
4.         if err != nil{
5.                 log.Fatalf(err)
6.         }
7.         defer conn.Close()
9.         // 创建一个通道
10.         ch, err := conn.Channel()
11.         if err != nil{
12.                 log.Fatalf(err)
13.         }
14.         defer ch.Close()
16.         // 声明一个交换机
17.         err = ch.ExchangeDeclare(
18.                 "logs_topic", // 交换机名称
19.                 "topic",      // 交换机类型
20.                 true,         // 是否持久化
21.                 false,        // 是否自动删除
22.                 false,        // 是否内部使用
23.                 false,        // 是否等待服务器响应
24.                 nil,          // 其他属性
25.         )
26.         if err != nil{
27.                 log.Fatalf(err)
28.         }
30.         // 声明一个临时队列
31.         q, err := ch.QueueDeclare(
32.                 "",    // 队列名称，留空表示由RabbitMQ自动生成
33.                 false, // 是否持久化
34.                 false, // 是否自动删除（当没有任何消费者连接时）
35.                 true,  // 是否排他队列（仅限于当前连接）
36.                 false, // 是否等待服务器响应
37.                 nil,   // 其他属性
38.         )
39.         if err != nil{
40.                 log.Fatalf(err)
41.         }
43.         // 将队列绑定到交换机上，并指定要接收的路由键
44.         err = ch.QueueBind(
45.                 q.Name,       // 队列名称
46.                 "example.#",  // 路由键，可以使用通配符*匹配一个单词
47.                 "logs_topic", // 交换机名称
48.                 false,        // 是否等待服务器响应
49.                 nil,          // 其他属性
50.         )
51.         if err != nil{
52.                 log.Fatalf(err)
53.         }
54.         // 创建一个消费者通道
55.         msgs, err := ch.Consume(
56.                 q.Name, // 队列名称
57.                 "",     // 消费者标识符，留空表示由RabbitMQ自动生成
58.                 true,   // 是否自动应答
59.                 false,  // 是否排他消费者
60.                 false,  // 是否阻塞
61.                 false,  // 是否等待服务器响应
62.                 nil,    // 其他属性
63.         )
64.         if err != nil{
65.                 log.Fatalf(err)
66.         }
67.         // 接收和处理消息
68.         forever := make(chan bool)
70.         go func() {
71.                 for d := range msgs {
72.                         log.Printf("Received a message: %s", d.Body)
73.                 }
74.         }()
76.         log.Printf("Waiting for messages...")
77.         // 阻塞
78.         <-forever
79. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4