六种工作模式介绍

1.简单(Simple)模式

P:生产者,也就是要发送消息的程序。

C:消费者:消息的接收者,会一直等待消息到来。

queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。

简单模式就是单发单收,消息的消费者监听消息队列,如果队列中有消息,就消费掉,消息被拿走后,自动从队列中删除。

2.工作队列(Work Queue)模式

这种模式就是多个消费者消费同一个队列中的消息,既然消费者多了那么就出现了消息分配的问题,所以对应着两种分配策略:

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

3.发布/订阅(Pub/Sub)模式

在这种模型中,多了一个 Exchange 角色,而且过程略有变化:

P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X (交换机)。

C:消费者,消息的接收者,会一直等待消息到来。

Queue:消息队列,接收消息、缓存消息。

Exchange:交换机(X) ,一方面,接收生产者发送的消息。另一方面,如何处理消息,递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。 Exchange有以下4种类型:

routing keyrouting keyrouting key

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

4.路由(Routing)模式

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Direct类型。在这种模式下:

routing key
routing keyrouting key
routing key
routing key
routing key
routing key

5.通配符(Tpoic)模式

路由模式其实就是上述发布/订阅模式的交换机转发类型变成了Topic类型。在这种模式下:

routing keyrouting key
*
#
routing keya.orange.brouting keyLazy.a.b.c

Go语言的实现

安装操作库

安装API库

streadway/amqp

go get github.com/streadway/amqp

封装rabbitmq

streadway/amqprabbitmq.go

简单(Simple)模式

生产者

消费者

运行结果:

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)模式

公平分发模式:

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

生产者

消费者1

消费者2

运行结果:

# 消费者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
Ack(false)

生产者

消费端限流:

实现公平派遣模式我们需要设置消费者端一次只能消费一条消息,之前我们已经进行了封装,直接在消费者端调用即可:

消费者1

消费者2

运行结果:

# 消费者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)模式

生产者

消费者1

消费者2

运行结果:

# 消费者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)模式

生产者

消费者1

消费者2

运行结果:

# 消费者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)模式

生产者

消费者1

消费者2

运行结果:

# 消费者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"