AI 助理
备案控制台
开发者社区 微服务 文章 正文

AMQP-RabbitMQ/3/发布订阅模式

2019-06-08 1193
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
函数计算FC，每月15万CU 3个月
简介： 3. 发布订阅模式 Publish/Subscribe - 全集监听fanout一次向多个消费者发送消息图示个人理解生产者定义Exchange，同时将Exchange的类型定义为fanout，并向该Exchange发送消息。

3. 发布订阅模式 Publish/Subscribe - 全集监听fanout

一次向多个消费者发送消息

  • 图示
    发布订阅模式

个人理解

  • 生产者定义Exchange，同时将Exchange的类型定义为fanout，并向该Exchange发送消息。
  • 消费者定义队列Queue，并将队列与该交换机进行绑定。之后交换机付负责将消息全量推送给每一个与之绑定的Queue

**RabbitMQ中消息传递模型的核心思想是生产者永远不会将任何消息直接发送到队列。实际上，生产者通常甚至不知道消息是否会被传递到任何队列。
相反，生产者只能向Exchange发送消息。Exchange所做的工作非常简单。一方面，它接收来自生产者的消息，另一方面将它们推送到队列。Exchange必须确切知道如何处理它收到的消息。它应该附加到特定队列吗？它应该附加到多个队列吗？或者它应该被丢弃。其规则由交换类型定义 。**
有几种交换类型可供选择：direct ,topic ,headers andfanout

  • fanout: 将它接收到的消息广播到所有绑定到它的消息队列上。(忽略路由键routingKey)
  • 生产者 - 发布者
package com.futao.springmvcdemo.mq.rabbit.ps;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.Cleanup;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 发布订阅-发布者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class Publisher {
    @SneakyThrows
    public static void main(String[] args) {
        @Cleanup
        Connection connection = RabbitMqConnectionTools.getConnection();
        @Cleanup
        Channel channel = connection.createChannel();
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.FANOUT.getExchangeName(), BuiltinExchangeType.FANOUT);
        String msg = "Hello RabbitMq!";
        for (int i = 0; i < 20; i++) {
            channel.basicPublish(ExchangeTypeEnum.FANOUT.getExchangeName(), "", null, (msg + i).getBytes());
            log.info("Send msg:[{}] success", (msg + i));
        }
    }
}
  • 消费者1 - 订阅者1
package com.futao.springmvcdemo.mq.rabbit.ps;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 发布订阅-订阅者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class SubscriberOne {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化队列
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_ONE.getQueueName(), durable, false, false, null);
        //定义交换器类型
        channel.exchangeDeclare(ExchangeTypeEnum.FANOUT.getExchangeName(), BuiltinExchangeType.FANOUT);
        //将消息队列与Exchange交换器绑定
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_ONE.getQueueName(), ExchangeTypeEnum.FANOUT.getExchangeName(), "");
        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_ONE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}
  • 消费者2 - 订阅者2
package com.futao.springmvcdemo.mq.rabbit.ps;

import com.futao.springmvcdemo.mq.rabbit.ExchangeTypeEnum;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 发布订阅-订阅者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class SubscriberTwo {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化队列
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_TWO.getQueueName(), durable, false, false, null);
        //定义交换器类型为fanout
        channel.exchangeDeclare(ExchangeTypeEnum.FANOUT.getExchangeName(), BuiltinExchangeType.FANOUT);
        //将消息队列与Exchange交换器进行绑定
        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_TWO.getQueueName(), ExchangeTypeEnum.FANOUT.getExchangeName(), "");
        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(2000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_TWO.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}

  • 注意

    • 没有在发布者定义队列，而是定义了交换器Exchange。发布者将消息发送到Exchange，而不是Queue
    • 在订阅者端，每个订阅者定义了自己的消息队列，并且将自己的消息队列与Exchange进行绑定。则在每次发布者向相应的Exchange发送消息的时候，Exchange会将消息发送至订阅了该Exchange的队列。(即：每个订阅者收到的消息都是一样的)
  • 测试结果
>>> 订阅者1
[main] INFO mq.rabbit.ps.SubscriberOne - Waiting for message...
[pool-1-thread-4] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!0],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-5] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!1],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-6] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!2],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-7] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!3],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-8] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!4],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-9] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!5],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-10] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!17],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-22] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!18],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]
[pool-1-thread-23] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!19],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]


>>> 订阅者2
[main] INFO mq.rabbit.ps.SubscriberTwo - Waiting for message...
[pool-1-thread-4] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!0],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-5] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!1],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-6] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!2],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-7] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!3],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-8] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!4],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-9] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!5],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-10] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!17],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-22] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!18],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
[pool-1-thread-23] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!19],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]
文章标签：
云消息队列 MQ
函数计算
消息中间件
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
futaosmile
目录
相关文章
忆~遂愿
|
消息中间件 存储 缓存
服务异步通信--RabbitMQ
服务异步通信--RabbitMQ
忆~遂愿
88 0
阿兵云原生
|
消息中间件 存储 网络协议
我们一起来学RabbitMQ 二：RabbiMQ 的 6 种模式的基本应用
我们一起来学RabbitMQ 二：RabbiMQ 的 6 种模式的基本应用
阿兵云原生
84 0
小宋1021
|
3月前
|
消息中间件 Kafka 数据安全/隐私保护
RabbitMQ异步通信详解
RabbitMQ异步通信详解
小宋1021
110 16
过。
|
7月前
|
消息中间件 缓存 中间件
服务器的异步通信——RabbitMQ1
服务器的异步通信——RabbitMQ
过。
64 0
过。
|
7月前
|
消息中间件 存储 JSON
服务器的异步通信——RabbitMQ2
服务器的异步通信——RabbitMQ
过。
59 0
王卫
|
消息中间件 存储 Java
【RabbitMQ四】——RabbitMQ发布订阅模式（Publish/Subscribe）
【RabbitMQ四】——RabbitMQ发布订阅模式（Publish/Subscribe）
王卫
361 1
讓丄帝愛伱
|
消息中间件 缓存 网络协议
RabbitMQ消息确认
RabbitMQ消息确认
讓丄帝愛伱
131 0
王卫
|
消息中间件 应用服务中间件 nginx
【RabbitMQ六】——RabbitMQ主题模式（Topic）
【RabbitMQ六】——RabbitMQ主题模式（Topic）
王卫
449 1
Withcomb
|
消息中间件
RabbitMq消息确认机制
RabbitMq消息确认机制
Withcomb
114 0
愿天堂没有BUG（公众号同名）
|
消息中间件 设计模式 网络协议
RabbitMQ初识以及简单模式初步
MQ是Message Queue的缩写，也就是消息队列。 MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的一种数据结构。指把要传输的数据（消息）放在队列中，用队列机制来实现消息传递——生产者产生消息并把消息放入队列，然后由消费者去处理。消费者可以到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给其推送消息。 消息队列这种技术主要用在分布式设计当中，其实可以说是一种设计模式。是相对同步系统而言的。同步系统是什么呢？ 同步，是当所有的操作都完毕，才会返回结果，比如用户支付，如果是同步的话就是当用户所发起的支付操作，然后只有等支付的这个业务成功然后才给用户返回结果说支付
愿天堂没有BUG（公众号同名）
91 0

热门文章

最新文章

  • 1
    《Stata统计分析与应用（第2版）》一3.4 图形的保存、合并及修改
  • 2
    Maven 3.6.3 的下载、安装、配置、检测（详细讲解）
  • 3
    基于区块链的机器学习模型创建方案
  • 4
    WPF listview item mouse enter/over popup
  • 5
    DevExpress 10.2.4系列-ASPxListBox根据列值不同设置不同图标
  • 6
    [Spring]支持注解的Spring调度器
  • 7
    红帽7.4新增的安全特性
  • 8
    hdu 1800 Flying to the Mars
  • 9
    研究 flex 4 开发后台管理项目（1）
  • 10
    Error: XML response that was returned...
  • 1
    《docker高级篇（大厂进阶）：2.DockerFile解析》包括：是什么、DockerFile构建过程解析、DockerFile常用保留字指令、案例、小总结
    112
  • 2
    《Java 人工智能开发：Eclipse 与 IntelliJ IDEA 双雄助力》
    42
  • 3
    《Java 在开源 AI 框架中的中流砥柱角色》
    30
  • 4
    《Java 与大数据框架：AI 数据预处理的强力联盟》
    33
  • 5
    《Java 优化秘籍：计算密集型 AI 任务加速指南》
    33
  • 6
    《Java 与 OpenAI 协同：开启智能编程新范式》
    24
  • 7
    uniapp开发鸿蒙，是前端新出路吗？
    28
  • 8
    鸿蒙(HarmonyOS)原生AI能力之文本识别
    36
  • 9
    鸿蒙开发之PixelMap介绍与实现图片变换
    33
  • 10
    鸿蒙(HarmonyOS)开发之无须申请权限写入图片到相册
    37

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列RabbitMQ入门课程
  • 带您快速了解阿里云消息队列RocketMQ 5.0版
  • RocketMQ 5.0学习资料
  • 分布式消息系统 Kafka 快速入门
  • 消息队列 RocketMQ 全类型业务消息学习课程

    • 相关电子书

    更多
  • 消息队列kafka介绍
  • 消息队列 Kafka 版差异化特性
  • 云原生消息队列Apache RocketMQ

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 部署RabbitMQ
  • RocketMQ中使用Java客户端发送消息和消费的应用
    • 下一篇
    【案例实战】SpringBoot整合阿里云文件上传OSS