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

陌无崖 · · 124 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

导语

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

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

特性

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

生产者

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

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

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

消费者

申请交换器

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

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

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

q, err := ch.QueueDeclare(
    "",    //name
    false, //durable
    false, //delete when usused
    true,  // exclusive
    false, //no-wait
    nil,   // arguments
)
队列绑定

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

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

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


image

image

直连交换器(路由模式)

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

生产者

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

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

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

消费者

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

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

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

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


image

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


image

推荐阅读


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


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

本文来自:简书

感谢作者:陌无崖

查看原文:RabbitMQ系列笔记广播模式和路由模式

入群交流(和以上内容无关):加入Go大咖交流群,免费领全套学习资料或添加微信:muxilin131420 备注:入群;或加QQ群:729884609

124 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传