0
0

二、RabbitMQ的五种工作模式

文章目录

六种工作模式介绍

1.简单（Simple）模式

在这里插入图片描述
P：生产者，也就是要发送消息的程序。
C：消费者：消息的接收者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

简单模式就是

单发单收

，消息的消费者监听消息队列，如果队列中有消息，就消费掉，消息被拿走后，自动从队列中删除。

2.工作队列（Work Queue）模式

在这里插入图片描述
这种模式就是多个消费者消费同一个队列中的消息，既然消费者多了那么就出现了消息分配的问题，所以对应着两种分配策略：

  1. 公平分发：每个消费者接收消息的概率是相等的，消息队列会循环依次给每个消费者发送消息，这种是默认的策略。
  2. 公平派遣：保证消费者在消费完某个消息，并发送确认信息后，消息队列才会向它推送新的消息，在此之间若是有新的消息，将会被推送到其它消费者，若所有的消费者都在消费消息，那么就会等待。

3.发布/订阅（Pub/Sub）模式

请添加图片描述
在这种模型中，多了一个 Exchange 角色，而且过程略有变化：

P：生产者,也就是要发送消息的程序，但是不再发送到队列中，而是发给X (交换机)。
C：消费者，消息的接收者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机(X) ，一方面，接收生产者发送的消息。另一方面，如何处理消息，递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。 Exchange有以下4种类型：

  1. Fanout：广播，将消息交给所有绑定到交换机的队列。
  2. Direct：全值匹配，把消息交给符合指定routing key的队列。
  3. Topic：通配符，与Direct类型类似，但Direct类型要求routing key完全相等，而Topic类型是对routing key进行模糊匹配，比Direct灵活。
  4. Headers：根据Message的一些头部信息来分发过滤Message，用的比较少。

注意：Exchange负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失。

4.路由（Routing）模式

在这里插入图片描述

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Direct类型。在这种模式下：
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的

routing key

进行判断，只有队列的

routing key

与消息的

routing key

完全一致，才会接收到消息。

P：生产者，向 Exchange 发送消息，发送消息时，会指定一个

routing key

。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与

routing key

完全匹配的队列。
C1：消费者，其所在队列指定了需要

routing key

为error的消息。
C2：消费者，其所在队列指定了需要

routing key

为 info、error、warning 的消息。

5.通配符（Tpoic）模式

在这里插入图片描述
路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Topic类型。在这种模式下：

队列的

routing key

与消息的

routing key

符合匹配规则，就可以接收到消息，有两种规则：

*

：可以（只能）匹配一个单词。

#

：可以匹配多个单词（或者零个）。

所以图中，

routing key

为

a.orange.b

的消息就会被转发到Q1，而

routing key

为

Lazy.a.b.c

的消息就会被发送到Q2。

Go语言的实现

安装操作库

安装API库

Go可以使用

streadway/amqp

库来操作rabbit，使用以下命令来安装：

go get github.com/streadway/amqp

封装rabbitmq

接下来我们对

streadway/amqp

库的内容进行一个二次封装，封装为一个

rabbitmq.go

文件：

package rabbitmq

import("encoding/json""github.com/streadway/amqp""log")// RabbitMQ RabbitMQ结构type RabbitMQ struct{
    channel  *amqp.Channel
    Name     string
    exchange string}// Connect 连接服务器funcConnect(s string)*RabbitMQ {//连接rabbitmq
    conn, e := amqp.Dial(s)failOnError(e,"连接Rabbitmq服务器失败！")
    ch, e := conn.Channel()failOnError(e,"无法打开频道！")
    mq :=new(RabbitMQ)
    mq.channel = ch
    return mq
}// New 初始化消息队列//第一个参数：rabbitmq服务器的链接，第二个参数：队列名字funcNew(s string, name string)*RabbitMQ {//连接rabbitmq
    conn, e := amqp.Dial(s)failOnError(e,"连接Rabbitmq服务器失败！")
    ch, e := conn.Channel()failOnError(e,"无法打开频道！")
    q, e := ch.QueueDeclare(
        name,//队列名false,//是否开启持久化true,//不使用时删除false,//排他false,//不等待nil,//参数)failOnError(e,"初始化消息队列失败！")

    mq :=new(RabbitMQ)
    mq.channel = ch
    mq.Name = q.Name
    return mq
}// QueueDeclare 声明queuefunc(q *RabbitMQ)QueueDeclare(queue string){_, e := q.channel.QueueDeclare(queue,false,true,false,false,nil)failOnError(e,"声明queue失败！")}// QueueDelete 删除queuefunc(q *RabbitMQ)QueueDelete(queue string){_, e := q.channel.QueueDelete(queue,false,true,false)failOnError(e,"删除queue失败！")}// Qos 配置queue参数func(q *RabbitMQ)Qos(){
    e := q.channel.Qos(1,0,false)failOnError(e,"无法设置QoS")}// NewExchange 初始化交换机//第一个参数：rabbitmq服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型funcNewExchange(s string, name string, typename string){//连接rabbitmq
    conn, e := amqp.Dial(s)failOnError(e,"连接Rabbitmq服务器失败！")
    ch, e := conn.Channel()failOnError(e,"无法打开频道！")
    e = ch.ExchangeDeclare(
        name,// name
        typename,// typetrue,// durablefalse,// auto-deletedfalse,// internalfalse,// no-waitnil,// arguments)failOnError(e,"初始化交换机失败！")}// ExchangeDelete 删除交换机func(q *RabbitMQ)ExchangeDelete(exchange string){
    e := q.channel.ExchangeDelete(exchange,false,true)failOnError(e,"删除交换机失败！")}// Bind 绑定消息队列到exchangefunc(q *RabbitMQ)Bind(exchange string, key string){
    e := q.channel.QueueBind(
        q.Name,
        key,
        exchange,false,nil,)failOnError(e,"绑定队列失败！")
    q.exchange = exchange
}// Send 向消息队列发送消息//Send方法可以往某个消息队列发送消息func(q *RabbitMQ)Send(queue string, body interface{}){
    str, e := json.Marshal(body)failOnError(e,"消息序列化失败！")
    e = q.channel.Publish("",//交换
        queue,//路由键false,//必填false,//立即
        amqp.Publishing{
            ReplyTo: q.Name,
            Body:[]byte(str),})
    msg :="向队列:"+ q.Name +"发送消息失败！"failOnError(e, msg)}// Publish 向exchange发送消息//Publish方法可以往某个exchange发送消息func(q *RabbitMQ)Publish(exchange string, body interface{}, key string){
    str, e := json.Marshal(body)failOnError(e,"消息序列化失败！")
    e = q.channel.Publish(
        exchange,
        key,false,false,
        amqp.Publishing{ReplyTo: q.Name,
            Body:[]byte(str)},)failOnError(e,"向交换机发送消息失败！")}// Consume 接收某个消息队列的消息func(q *RabbitMQ)Consume()<-chan amqp.Delivery {
    c, e := q.channel.Consume(
        q.Name,//指定从哪个队列中接收消息"",true,false,false,false,nil,)failOnError(e,"接收消息失败！")return c
}// Close 关闭队列连接func(q *RabbitMQ)Close(){
    q.channel.Close()}//错误处理函数funcfailOnError(err error, msg string){if err !=nil{
        log.Fatalf("%s: %s", msg, err)}}

简单（Simple）模式

生产者

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    producer := rabbitmq.New("amqp://guest:[email protected]:5672/","simple")

    i :=0for{// 每隔2s发送一次消息
        time.Sleep(time.Second *2)
        producer.Send("simple"," simple message: "+strconv.Itoa(i))
        i = i +1}}

消费者

funcmain(){
    consumer := rabbitmq.New("amqp://guest:[email protected]:5672/","simple")//接收消息时，指定
    messages := consumer.Consume()gofunc(){for ch :=range messages {
            log.Printf("Received a message: %s", ch.Body)// 消费消息要用3s
            time.Sleep(time.Second *3)}}()select{}}

运行结果：

2022/11/05 18:54:47 Received a message: " simple message: 0"
2022/11/05 18:54:52 Received a message: " simple message: 1"
2022/11/05 18:54:57 Received a message: " simple message: 2"

工作队列（Work Queue）模式

公平分发模式：

公平分发模式采用的是轮询机制，它会将数个任务按顺序平均分发给消费者。

生产者

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    producer := rabbitmq.New("amqp://guest:[email protected]:5672/","worker")

    i :=0for{// 每隔2s发送一次消息
        time.Sleep(time.Second *2)
        producer.Send("worker"," worker message: "+strconv.Itoa(i))
        i = i +1}}

消费者1

funcmain(){
    consumer1 := rabbitmq.New("amqp://guest:[email protected]:5672/","worker")//接收消息
    messages := consumer1.Consume()gofunc(){for ch :=range messages {
            log.Printf("Received a message: %s", ch.Body)// 消费消息要用3s
            time.Sleep(time.Second *3)}}()select{}}

消费者2

funcmain(){
    consumer2 := rabbitmq.New("amqp://guest:[email protected]:5672/","worker")//接收消息
    messages := consumer2.Consume()gofunc(){for ch :=range messages {
            log.Printf("Received a message: %s", ch.Body)// 消费消息要用3s
            time.Sleep(time.Second *3)}}()select{}}

运行结果：

# 消费者1
2022/11/05 19:45:03 Received a message: " worker message: 0"
2022/11/05 19:45:07 Received a message: " worker message: 2"
2022/11/05 19:45:11 Received a message: " worker message: 4"

# 消费者2
2022/11/05 19:45:05 Received a message: " worker message: 1"
2022/11/05 19:45:09 Received a message: " worker message: 3"
2022/11/05 19:45:13 Received a message: " worker message: 5"

可以发现，公平模式下，偶数消息都被发送给了消费者1，而奇数消息都被发送给了消费者2。

公平派遣模式：

有时候，如果消息之间的复杂度不同，那么不同消费者消费消息所用的时间会不同。这个时候如果使用公平派发模式，可能会造成某一个消费者需要消费的消息积压过多。可以采用公平派遣模式：

公平派遣模式下发送端与公平分发相同，消费者端只需要加一段配置代码，我们可以将预取计数设置为1。这告诉RabbitMQ一次不要给消费者一个以上的消息。换句话说，在处理并确认上一条消息之前，不要将新消息发送给消费者。而是将其分派给不忙的下一个消费者。

关于消息的确认：

为了确保消息永不丢失，RabbitMQ支持 消息确认。消费者发送回一个确认（acknowledgement），以告知RabbitMQ已经接收，处理了特定的消息，并且RabbitMQ可以自由删除它。

我们之前的代码中，RabbitMQ一旦向消费者传递了一条消息，便立即将其标记为删除（调用Consumer的第三个参数是autoAck，表示是否自动回复）。在这种情况下，如果你终止一个消费者那么你就可能会丢失这个任务，我们还将丢失所有已经交付给这个消费者的尚未消费的消息。如果一个消费者意外宕机了，那么我们希望将任务交付给其他消费者来消费者。

所以一旦向消费者传递了一条消息，就不能马上将其标记为删除，而是要手动确认。我们需要在创建消费者的时候将

autoAck

参数标记为false：

// Consume 接收某个消息队列的消息func(q *RabbitMQ)Consume()<-chan amqp.Delivery {
    c, e := q.channel.Consume(
        q.Name,//指定从哪个队列中接收消息"",false,// 不自动确认消息false,false,false,nil,)failOnError(e,"接收消息失败！")return c
}

然后每消费完一条消息需要调用

Ack(false)

函数手动回复。

生产者

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    producer := rabbitmq.New("amqp://guest:[email protected]:5672/","worker")

    i :=0for{// 每隔2s发送一次消息
        time.Sleep(time.Second *2)
        producer.Send("worker"," worker message: "+strconv.Itoa(i))
        i = i +1}}

消费端限流：
实现

公平派遣模式

我们需要设置消费者端一次只能消费一条消息，之前我们已经进行了封装，直接在消费者端调用即可：

// Qos 配置queue参数func(q *RabbitMQ)Qos(){
    e := q.channel.Qos(1,0,false)failOnError(e,"无法设置QoS")}

消费者1

funcmain(){
    consumer1 := rabbitmq.New("amqp://guest:[email protected]:5672/","worker")// 指定一次只消费一条消息，直到消费完才重新接收
    consumer1.Qos()//接收消息
    messages := consumer1.Consume()gofunc(){for ch :=range messages {
            log.Printf("Received a message: %s", ch.Body)// 消费消息要用10s
            time.Sleep(time.Second *10)// 手动回复
            ch.Ack(false)}}()select{}}

消费者2

funcmain(){
    consumer2 := rabbitmq.New("amqp://guest:[email protected]:5672/","worker")// 指定一次只消费一条消息，直到消费完才重新接收
    consumer2.Qos()//接收消息
    messages := consumer2.Consume()gofunc(){for ch :=range messages {
            log.Printf("Received a message: %s", ch.Body)// 消费消息要用2s
            time.Sleep(time.Second *2)// 手动回复
            ch.Ack(false)}}()select{}}

运行结果：

# 消费者1
2022/11/05 20:31:26 Received a message: " worker message: 0"
2022/11/05 20:31:36 Received a message: " worker message: 5"

# 消费者2
2022/11/05 20:31:28 Received a message: " worker message: 1"
2022/11/05 20:31:30 Received a message: " worker message: 2"
2022/11/05 20:31:32 Received a message: " worker message: 3"
2022/11/05 20:31:34 Received a message: " worker message: 4"
2022/11/05 20:31:38 Received a message: " worker message: 6"
2022/11/05 20:31:40 Received a message: " worker message: 7"

发布/订阅（Pub/Sub）模式

生产者

funcmain(){
    producer := rabbitmq.New("amqp://guest:[email protected]:5672/","queue")
    rabbitmq.NewExchange("amqp://guest:[email protected]:5672/","exchange1","fanout")
    i :=0for{
        time.Sleep(time.Second)// fanout模式下不用routing key
        producer.Publish("exchange1","pubsub message: "+strconv.Itoa(i),"")
        i = i +1}}

消费者1

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    consumer1 := rabbitmq.New("amqp://guest:[email protected]:5672/","queue1")// 队列绑定到exchange
    consumer1.Bind("exchange1","")//接收消息
    msgs := consumer1.Consume()gofunc(){for d :=range msgs {
            log.Printf("Consumer1 received a message: %s", d.Body)
            d.Ack(false)}}()select{}}

消费者2

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    consumer2 := rabbitmq.New("amqp://guest:[email protected]:5672/","queue2")// 队列绑定到exchange
    consumer2.Bind("exchange1","")//接收消息
    msgs := consumer2.Consume()gofunc(){for d :=range msgs {
            log.Printf("Consumer2 received a message: %s", d.Body)
            d.Ack(false)}}()select{}}

运行结果：

# 消费者1
2022/11/05 22:32:19 Consumer1 received a message: "pubsub message: 0"
2022/11/05 22:32:20 Consumer1 received a message: "pubsub message: 1"
2022/11/05 22:32:21 Consumer1 received a message: "pubsub message: 2"
2022/11/05 22:32:22 Consumer1 received a message: "pubsub message: 3"
2022/11/05 22:32:23 Consumer1 received a message: "pubsub message: 4"
2022/11/05 22:32:24 Consumer1 received a message: "pubsub message: 5"

# 消费者2
2022/11/05 22:32:19 Consumer2 received a message: "pubsub message: 0"
2022/11/05 22:32:20 Consumer2 received a message: "pubsub message: 1"
2022/11/05 22:32:21 Consumer2 received a message: "pubsub message: 2"
2022/11/05 22:32:22 Consumer2 received a message: "pubsub message: 3"
2022/11/05 22:32:23 Consumer2 received a message: "pubsub message: 4"
2022/11/05 22:32:24 Consumer2 received a message: "pubsub message: 5"

路由（Routing）模式

生产者

funcmain(){
    producer := rabbitmq.New("amqp://guest:[email protected]:5672/","queue")// 指定为direct类型
    rabbitmq.NewExchange("amqp://guest:[email protected]:5672/","exchange","direct")
    i :=0for{
        time.Sleep(time.Second)// 如果是奇数，就发key1// 如果是偶数，就发key2if i%2!=0{
            producer.Publish("exchange","routing message: "+strconv.Itoa(i),"key1")}else{
            producer.Publish("exchange","routing message: "+strconv.Itoa(i),"key2")}
        i = i +1}}

消费者1

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    consumer1 := rabbitmq.New("amqp://guest:[email protected]:5672/","queue1")// 队列绑定到exchange
    consumer1.Bind("exchange","key1")//接收消息
    msgs := consumer1.Consume()gofunc(){for d :=range msgs {
            log.Printf("Consumer1 received a message: %s", d.Body)
            d.Ack(false)}}()select{}}

消费者2

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    consumer2 := rabbitmq.New("amqp://guest:[email protected]:5672/","queue2")// 队列绑定到exchange
    consumer2.Bind("exchange","key2")//接收消息
    msgs := consumer2.Consume()gofunc(){for d :=range msgs {
            log.Printf("Consumer2 received a message: %s", d.Body)
            d.Ack(false)}}()select{}}

运行结果：

# 消费者1
2022/11/05 22:51:10 Consumer1 received a message: "routing message: 1"
2022/11/05 22:51:12 Consumer1 received a message: "routing message: 3"
2022/11/05 22:51:14 Consumer1 received a message: "routing message: 5"
2022/11/05 22:51:16 Consumer1 received a message: "routing message: 7"

# 消费者2
2022/11/05 22:51:11 Consumer2 received a message: "routing message: 0"
2022/11/05 22:51:13 Consumer2 received a message: "routing message: 2"
2022/11/05 22:51:15 Consumer2 received a message: "routing message: 4"
2022/11/05 22:51:17 Consumer2 received a message: "routing message: 6"

通配符（Tpoic）模式

生产者

funcmain(){
    producer := rabbitmq.New("amqp://guest:[email protected]:5672/","queue")// 指定为topic类型
    rabbitmq.NewExchange("amqp://guest:[email protected]:5672/","exchange2","topic")var i intfor{
        time.Sleep(time.Second)if i%2!=0{
            producer.Publish("exchange2","topic message: "+strconv.Itoa(i),"a.test.b.c")}else{
            producer.Publish("exchange2","topic message: "+strconv.Itoa(i),"a.test.b")}
        i++}}

消费者1

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    consumer1 := rabbitmq.New("amqp://guest:[email protected]:5672/","queue1")// 队列绑定到exchange
    consumer1.Bind("exchange2","*.test.*")//接收消息
    msgs := consumer1.Consume()gofunc(){for d :=range msgs {
            log.Printf("Consumer1 received a message: %s", d.Body)
            d.Ack(false)}}()select{}}

消费者2

funcmain(){//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
    consumer2 := rabbitmq.New("amqp://guest:[email protected]:5672/","queue2")// 队列绑定到exchange
    consumer2.Bind("exchange2","#.test.#")//接收消息
    msgs := consumer2.Consume()gofunc(){for d :=range msgs {
            log.Printf("Consumer2 received a message: %s", d.Body)
            d.Ack(false)}}()select{}}

运行结果：

# 消费者1
2022/11/05 23:09:53 Consumer1 received a message: "topic message: 0"
2022/11/05 23:09:55 Consumer1 received a message: "topic message: 2"
2022/11/05 23:09:57 Consumer1 received a message: "topic message: 4"
2022/11/05 23:09:59 Consumer1 received a message: "topic message: 6"

# 消费者2
2022/11/05 23:09:53 Consumer2 received a message: "topic message: 0"
2022/11/05 23:09:54 Consumer2 received a message: "topic message: 1"
2022/11/05 23:09:55 Consumer2 received a message: "topic message: 2"
2022/11/05 23:09:56 Consumer2 received a message: "topic message: 3"
2022/11/05 23:09:57 Consumer2 received a message: "topic message: 4"
2022/11/05 23:09:58 Consumer2 received a message: "topic message: 5"
2022/11/05 23:09:59 Consumer2 received a message: "topic message: 6"
标签： rabbitmq
本文转载自: https://blog.csdn.net/qq_49723651/article/details/127705928
版权归原作者 Onemorelight95 所有， 如有侵权，请联系我们删除。
登录后发布评论

“二、RabbitMQ的五种工作模式”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

【漏洞复现】LiveBos UploadFile 任意文件上传漏洞

针对一个红队病毒样本逆向分析

拯救者Y7000p 2024款安装ubuntu20.04无wifi问题

Docker本地部署Chatbot Ollama搭建AI聊天机器人并实现远程交互

2024 年可免费下载的 6 款最佳 iOS 解锁软件

Qt与MQTT交互通信

Android应用性能优化

文章导航