MQ的发布订阅模式（fanout）｜ RabbitMQ系列（七）

这是我参与8月更文挑战的第12天，活动详情查看：8月更文挑战markdown

RabbitMQ系列汇总：RabbitMQ系列工具

前言

回顾一下简单模式和工做模式post
- 简单模式：一个生产者对应一个消费者。
- 工做模式：一个生产者对应多个消费者。
如今来介绍一下交换机是干吗的测试
RabbitMQ 消息传递模型的核心思想是: 生产者生产的消息从不会直接发送到队列。实际上，一般生产者甚至都不知道这些消息传递传递到了哪些队列中。spa
相反，生产者只能将消息发送到交换机(exchange)，交换机工做的内容很是简单，一方面它接收来自生产者的消息，另外一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。code
是应该把这些消息放到特定队列仍是说把他们到许多队列中仍是说应该丢弃它们。这就的由交换机的类型来决定。orm
交换机共有四种类型队列
- 直接(direct)
- 主题(topic)
- 标题(headers)
- 扇出(fanout)
- 通常MQ会帮咱们建立一些默认的交换机，能够直接拿来使用，也能够本身建立不一样类型的交换机
这里须要讲下无名交换机，也就是默认交换机，先前的几篇文章咱们都没有指定交换机路由
- 第一个参数是交换机的名称。空字符串表示默认或无名称交换机：消息能路由发送到队列中其实是由 routingKey(bindingkey)绑定 key 指定的，若是它存在的话。
- 那么这种状况下，MQ会走默认的交换机（AMQP default）

还有东西叫临时队列字符串

//建立临时队列
String queueName = channel.queueDeclare().getQueue();
复制代码

直接执行确定报错的：明显告诉咱们队列名不要用amq来命名，因此能够将代码稍微改造下

channel.queueDeclare("dy"+queueName,false,false,false,null);
复制代码

了解了交换机以后，咱们就能够正式开始发布订阅模式啦~

1、生产者

尽可能在生产者指定交换机，毕竟按照正常逻辑来看，确定是先发送，才能接受。不然的话接受了个寂寞哦~

   public static void publishMessageIndividually() throws Exception {
        //工具类获取通道
        Channel channel = RabbitMqUtils.getChannel();
        /**
         * 指定交换机和模式
         * 参数一：指定的交换机名称
         * 参数二：指定的交换机模式
         */
        channel.exchangeDeclare(ChangeNameConstant.FANOUT_MODEL,"fanout");

        Scanner sc = new Scanner(System.in);
        System.out.println("请输入信息");
        while (sc.hasNext()) {
            String message = sc.nextLine();
            channel.basicPublish(ChangeNameConstant.FANOUT_MODEL, "20210804", null, message.getBytes("UTF-8"));
            System.out.println("生产者发出消息" + message);
        }
    }
复制代码

因而可知
- 建立了一个生产者
- 创建了一个交换机Name= fanout_pattern
- 交换机的类型是 fanout（扇型）

2、消费者

消费者A

/**
 * 这是一个测试的消费者
 *@author DingYongJun
 *@date 2021/8/1
 */
public class DyConsumerTest_Fanout01 {

    public static void main(String[] args) throws Exception{
        //使用工具类来建立通道
        Channel channel = RabbitMqUtils.getChannel();

        /**
         * 生成一个临时的队列 队列的名称是随机的
         * 当消费者断开和该队列的链接时 队列自动删除
         */
        String queueName = channel.queueDeclare().getQueue();
        //把该临时队列绑定咱们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
        channel.queueBind(queueName, ChangeNameConstant.FANOUT_MODEL, "20210804");
        System.out.println("交换机A等待接收消息,把接收到的消息打印在屏幕.....");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("控制台打印接收到的消息"+message);
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
            System.out.println("消息中断了~");
        });
    }
}
复制代码

消费者B代码同样，只是改为消费者B等待接受消息~
两个消费者的申明的临时队列成功绑定到了交换机上！
执行结果
- 生产者
- 消费者A
- 消费者B
生产者给交换机发了一条消息。
两个消费者都收到了消息。

3、总结

很明显，这个地方咱们不会将消息直接发到队列了
而是将消息发送给交换机
交换机寻找哪些队列绑定了本交换机
而后将消息发送至全部被绑定的队列
最终队列发送消息给消费者

路漫漫其修远兮，吾必将上下求索~

若是你认为i博主写的不错！写做不易，请点赞、关注、评论给博主一个鼓励吧~hahah

MQ的发布订阅模式（fanout）｜ RabbitMQ系列（七）

相关文章

前言

1、生产者

2、消费者

3、总结