文章目录
六种工作模式介绍Go语言的实现安装操作库简单（Simple）模式工作队列（Work Queue）模式发布/订阅（Pub/Sub）模式路由（Routing）模式通配符（Tpoic）模式
六种工作模式介绍
1.简单（Simple）模式
P：生产者，也就是要发送消息的程序。 C：消费者：消息的接收者，会一直等待消息到来。 queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
简单模式就是单发单收，消息的消费者监听消息队列，如果队列中有消息，就消费掉，消息被拿走后，自动从队列中删除。
2.工作队列（Work Queue）模式
这种模式就是多个消费者消费同一个队列中的消息，既然消费者多了那么就出现了消息分配的问题，所以对应着两种分配策略：
公平分发：每个消费者接收消息的概率是相等的，消息队列会循环依次给每个消费者发送消息，这种是默认的策略。公平派遣：保证消费者在消费完某个消息，并发送确认信息后，消息队列才会向它推送新的消息，在此之间若是有新的消息，将会被推送到其它消费者，若所有的消费者都在消费消息，那么就会等待。
3.发布/订阅（Pub/Sub）模式
在这种模型中，多了一个 Exchange 角色，而且过程略有变化：
P：生产者,也就是要发送消息的程序，但是不再发送到队列中，而是发给X (交换机)。 C：消费者，消息的接收者，会一直等待消息到来。 Queue：消息队列，接收消息、缓存消息。 Exchange：交换机(X) ，一方面，接收生产者发送的消息。另一方面，如何处理消息，递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。 Exchange有以下4种类型：
Fanout：广播，将消息交给所有绑定到交换机的队列。Direct：全值匹配，把消息交给符合指定routing key的队列。Topic：通配符，与Direct类型类似，但Direct类型要求routing key完全相等，而Topic类型是对routing key进行模糊匹配，比Direct灵活。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 连接服务器
func Connect(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服务器的链接，第二个参数：队列名字
func New(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 声明queue
func (q *RabbitMQ) QueueDeclare(queue string) {
_, e := q.channel.QueueDeclare(queue, false, true, false, false, nil)
failOnError(e, "声明queue失败！")
}
// QueueDelete 删除queue
func (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服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型
func NewExchange(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, // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
failOnError(e, "初始化交换机失败！")
}
// ExchangeDelete 删除交换机
func (q *RabbitMQ) ExchangeDelete(exchange string) {
e := q.channel.ExchangeDelete(exchange, false, true)
failOnError(e, "删除交换机失败！")
}
// Bind 绑定消息队列到exchange
func (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()
}
//错误处理函数
func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
}
}
简单（Simple）模式
生产者
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
i := 0
for {
// 每隔2s发送一次消息
time.Sleep(time.Second * 2)
producer.Send("simple", " simple message: "+strconv.Itoa(i))
i = i + 1
}
}
消费者
func main() {
consumer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
//接收消息时，指定
messages := consumer.Consume()
go func() {
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）模式
公平分发模式：
公平分发模式采用的是轮询机制，它会将数个任务按顺序平均分发给消费者。
生产者
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
i := 0
for {
// 每隔2s发送一次消息
time.Sleep(time.Second * 2)
producer.Send("worker", " worker message: "+strconv.Itoa(i))
i = i + 1
}
}
消费者1
func main() {
consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
//接收消息
messages := consumer1.Consume()
go func() {
for ch := range messages {
log.Printf("Received a message: %s", ch.Body)
// 消费消息要用3s
time.Sleep(time.Second * 3)
}
}()
select {}
}
消费者2
func main() {
consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
//接收消息
messages := consumer2.Consume()
go func() {
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)函数手动回复。
生产者
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
i := 0
for {
// 每隔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
func main() {
consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
// 指定一次只消费一条消息，直到消费完才重新接收
consumer1.Qos()
//接收消息
messages := consumer1.Consume()
go func() {
for ch := range messages {
log.Printf("Received a message: %s", ch.Body)
// 消费消息要用10s
time.Sleep(time.Second * 10)
// 手动回复
ch.Ack(false)
}
}()
select {}
}
消费者2
func main() {
consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
// 指定一次只消费一条消息，直到消费完才重新接收
consumer2.Qos()
//接收消息
messages := consumer2.Consume()
go func() {
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）模式
生产者
func main() {
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange1", "fanout")
i := 0
for {
time.Sleep(time.Second)
// fanout模式下不用routing key
producer.Publish("exchange1", "pubsub message: "+strconv.Itoa(i), "")
i = i + 1
}
}
消费者1
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
// 队列绑定到exchange
consumer1.Bind("exchange1", "")
//接收消息
msgs := consumer1.Consume()
go func() {
for d := range msgs {
log.Printf("Consumer1 received a message: %s", d.Body)
d.Ack(false)
}
}()
select {}
}
消费者2
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
// 队列绑定到exchange
consumer2.Bind("exchange1", "")
//接收消息
msgs := consumer2.Consume()
go func() {
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）模式
生产者
func main() {
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
// 指定为direct类型
rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange", "direct")
i := 0
for {
time.Sleep(time.Second)
// 如果是奇数，就发key1
// 如果是偶数，就发key2
if 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
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
// 队列绑定到exchange
consumer1.Bind("exchange", "key1")
//接收消息
msgs := consumer1.Consume()
go func() {
for d := range msgs {
log.Printf("Consumer1 received a message: %s", d.Body)
d.Ack(false)
}
}()
select {}
}
消费者2
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
// 队列绑定到exchange
consumer2.Bind("exchange", "key2")
//接收消息
msgs := consumer2.Consume()
go func() {
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）模式
生产者
func main() {
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
// 指定为topic类型
rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange2", "topic")
var i int
for {
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
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
// 队列绑定到exchange
consumer1.Bind("exchange2", "*.test.*")
//接收消息
msgs := consumer1.Consume()
go func() {
for d := range msgs {
log.Printf("Consumer1 received a message: %s", d.Body)
d.Ack(false)
}
}()
select {}
}
消费者2
func main() {
//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
// 队列绑定到exchange
consumer2.Bind("exchange2", "#.test.#")
//接收消息
msgs := consumer2.Consume()
go func() {
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"
精彩内容
您阅读本篇文章共花了：
发表评论