RabbitMQ系列笔记广播模式和路由模式

本文已被阅读过 Posted by 陌无崖 on 2019-08-07

导语

上一节介绍了简单的工作模式,即一个队列可以被多个消费者进行消费,只有一条消息被送到消费者,采用公平调度的方式,在以往的例子中似乎我们还没用到交换器进行发送消息,我们都知道,往队列里发送消息,是需要用交换器进行分发的消息的,为什么我们没有申请交换器仍然可以发送消息呢?因为在RabbitMQ服务器中,如果不申请交换器,服务器会使用默认的交换器,所以说,交换器在发送消息的时候必不可少,今天我们学习两种交换器分别为fanout(扇形交换器)和direct(直连交换器)

扇形交换器(日志记录器)

特性

分发消息到与之绑定的所有队列,发送消息到交换器时会忽略routing key(路由),我们可以称它为广播消息。
####生产者

申请交换器
  • 指定交换器的名字
  • 指定交换器的类型
1
2
3
4
5
6
7
8
9
err = ch.ExchangeDeclare(
"logs", //交换器的名字
"fanout", //交换器的类型、这里为广播类型
true, //是否持久
false, //无用的时候是否自动删除
false, //true表示是。客户端无法直接发送msg到内部交换器,只有交换器可以发送msg到内部交换器。
false, //no-wait
nil, //arguments
)
发送消息

生产者制制定了交换器后就可以直接发送消息了

1
2
3
4
5
6
7
8
9
10
11
err = ch.Publish(
"logs", //发送到交换机的名字
"", // 路由,即队列的名字
false, //mandatory
false, //immediate
amqp.Publishing{
DeliveryMode: amqp.Persistent, //消息的持久化
ContentType: "text/plain",
Body: []byte(body),
},
)

消费者

申请交换器

消费者申请交换器需要和生产者完全保持一致。

1
2
3
4
5
6
7
8
9
err = ch.ExchangeDeclare(
"logs", //交换器的名字
"fanout", //交换器的类型、这里为广播类型
true, //是否持久
false, //无用的时候是否自动删除
false, //true表示是。客户端无法直接发送msg到内部交换器,只有交换器可以发送msg到内部交换器。
false, //no-wait
nil, //arguments
)
申请队列

在消费者申请队列时,因为这个案例负责记录日志,而日志可以从多个生产者进行产生,因此我们用一个临时队列存储,不需要进行持久化。

1
2
3
4
5
6
7
8
q, err := ch.QueueDeclare(
"", //name
false, //durable
false, //delete when usused
true, // exclusive
false, //no-wait
nil, // arguments
)
队列绑定

队列需要选择交换器,来收消息

1
2
3
4
5
6
7
err = ch.QueueBind(
q.Name, //队列的名字
"", //routing key
"logs", //所绑定的交换器
false,
nil,
)
运行

我们启动两个消费者一个生产者,生产者发送一条消息,这两个消费者会同时收到该消息

直连交换器(路由模式)

对于上面的案例,可能有的同学会想,我如果不想让所有的消费者收到消息,只想订阅我的消费者收到消息,订阅路由不同,消息会发送到不同的消费者不是更好吗,因此我们需要用direct交换器进行分发消息

生产者

申请交换器
1
2
3
4
5
6
7
8
9
err = ch.ExchangeDeclare(
"logs_direct", //交换器的名字
"direct", //交换器的类型、这里为广播类型
true, //是否持久
false, //无用的时候是否自动删除
false, //true表示是。客户端无法直接发送msg到内部交换器,只有交换器可以发送msg到内部交换器。
false, //no-wait
nil, //arguments
)
发送消息

生产者发送消息的时候需要指定路由

1
2
3
4
5
6
7
8
9
10
11
err = ch.Publish(
"logs_direct", //发送到交换机的名字
"info", // 路由,
false, //mandatory
false, //immediate
amqp.Publishing{
DeliveryMode: amqp.Persistent, //消息的持久化
ContentType: "text/plain",
Body: []byte(body),
},
)

消费者

申请交换器
1
2
3
4
5
6
7
8
9
err = ch.ExchangeDeclare(
"logs_direct", //交换器的名字
"direct", //交换器的类型、这里为广播类型
true, //是否持久
false, //无用的时候是否自动删除
false, //true表示是。客户端无法直接发送msg到内部交换器,只有交换器可以发送msg到内部交换器。
false, //no-wait
nil, //arguments
)
申请队列
1
2
3
4
5
6
7
8
q, err := ch.QueueDeclare(
"", //name
false, //durable
false, //delete when usused
true, // exclusive
false, //no-wait
nil, // arguments
)
队列绑定

治理需要注意的是,绑定时不仅要指定路由还要指定交换器

1
2
3
4
5
6
7
err = ch.QueueBind(
q.Name, //队列的名字
"info", //routing key
"logs_direct", //所绑定的交换器
false,
nil,
)
运行

为了更好的进行演示,我们需要开启3个消费者,在代码中只需要更改路由的名称,两个消费者的路由一样,另一个不一样,两个生产者发送消息,路由不同,我们会看到下面不同的控制台,有两个消费者一致的消息,另一个不一样。

当我们结束程序的时,最终看我们的可视化界面,我们的队列也被删除了

推荐阅读


本文欢迎转载,转载请联系作者,谢谢!


打开微信扫一扫,关注微信公众号