202、RabbitMQ 之 使用 fanout 类型的Exchange 实现 Pub-Sub 消息模型---fanout类型就是广播类型

目录

  • ★ 使用 fanout 类型的Exchange 实现 Pub-Sub 消息模型
  • 代码演示：
    • 生产者：producer
    • 消费者：Consumer01
    • 消费者：Consumer02
    • 测试结果
  • 完整代码
    • ConnectionUtil
    • Publisher
    • Consumer01
    • Consumer02
    • pom.xml

★ 使用 fanout 类型的Exchange 实现 Pub-Sub 消息模型

就是声明一个 fanout 类型的 Exchange 来分发消息。消费者进行消费
fanout 类型就是广播模式。

fanout 类型 的 Exchange 不会判断消息的路由key，直接将消息分发给绑定到该Exchange的所有队列。

生产者发送一条消息到fanout类型的Exchange后，绑定到该Exchange的所有队列都会收到该消息的一条副本，
而消费者也能分别从不同的队列中读取消息，互不干扰。

▲ fanout类型的Exchange可以很好地模拟JMS的Pub-Sub消息模型。

在这里插入图片描述

代码演示：

都是在前面一篇的代码基础上修改的。
需求：使用 fanout 类型的Exchange ，实行发布-订阅的功能，其实就是创建一个生产者和两个消费者，实现广播模式的消息分发。

在这里插入图片描述

生产者：producer

在生产者中声明Exchange ，然后声明两个消息队列 Queue，
然后给这个Exchange 绑定 这个两个Queue
在这里插入图片描述

在这里插入图片描述

消费者：Consumer01

两个消费者的代码没啥区别，
消费方法的参数 autoAck 都是true， 都是自动确认消费。
两个消费者各自消费自己指定的消息队列。

在这里插入图片描述

在这里插入图片描述

消费者：Consumer02

在这里插入图片描述
在这里插入图片描述

测试结果

消费生产者发送10条消息，两个消费者都能各自消费到10条消息就是正确的。

消息生产者使用fanout这个广播的类型发送消息。
在这里插入图片描述
两个消费者都能消费到10条消息，正确。
在这里插入图片描述

完整代码

ConnectionUtil

package cn.ljh.rabbitmq.util;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;//连接工具
public class ConnectionUtil
{//获取连接的方法public static Connection getConnection() throws IOException, TimeoutException{//创建连接工厂----这个ConnectionFactory源码可以看出有构造器，所以直接new一个出来ConnectionFactory connectionFactory =  new ConnectionFactory();//设置连接信息connectionFactory.setHost("localhost");connectionFactory.setPort(5672);connectionFactory.setUsername("ljh");connectionFactory.setPassword("123456");connectionFactory.setVirtualHost("/"); //连接虚拟主机//从连接工厂获取连接Connection connection = connectionFactory.newConnection();//返回连接return connection;}
}

Publisher

package cn.ljh.rabbitmq.producer;import cn.ljh.rabbitmq.consumer.Consumer01;
import cn.ljh.rabbitmq.consumer.Consumer02;
import cn.ljh.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;//消息生产者--使用fanout类型的exchange------就是广播模式
public class Publisher
{//常量：定义个Exchange的名字作为常量public static final String EXCHANGE_NAME = "myex01.fanout";public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接Connection conn = ConnectionUtil.getConnection();//2、通过Connection获取Channel。Channel channel = conn.createChannel();//3、调用exchangeDeclare()方法声明Exchange、调用queueDeclare()方法声明队列，并完成队列与Exchange的绑定channel.exchangeDeclare(EXCHANGE_NAME,/* Exchange名字 */BuiltinExchangeType.FANOUT,/* Exchange 类型 */true,/* 是否持久化 */false,/* 是否自动栅除 */false,/* 是否为内部的 Exchange */null /* 指定 Exchange 的额外属性 */);//声明多个消息队列------声明第1个消息队列channel.queueDeclare(Consumer01.QUEUE01, true, false, false, null);//把 Exchange 和 Queue 绑定起来，绑定第一个消息队列channel.queueBind(Consumer01.QUEUE01,EXCHANGE_NAME,"" /* 因为Exchange 是fanout类型，所以无需 路由key */,null /* 指定 Exchange 的额外属性 */);//声明第2个消息队列channel.queueDeclare(Consumer02.QUEUE02, true, false, false, null);//把 Exchange 和 Queue 绑定起来，绑定第2个消息队列channel.queueBind(Consumer02.QUEUE02,EXCHANGE_NAME,"" /* 因为Exchange 是fanout类型，所以无需 路由key */,null /* 指定 Exchange 的额外属性 */);//生产者发送10条消息for (int i = 1; i <= 10; i++){String message = "生产者发送的第【 " + i + " 】条消息的内容";//4、调用Channel的basicPublish()方法发送消息channel.basicPublish(EXCHANGE_NAME /* 向这个 fanout类型的 Exchange 发送消息 */,"" /* 因为 Exchange 是fanout 类型，所以有没有路由key都无所谓 */,null /*指定额外的消息的属性*/,message.getBytes(StandardCharsets.UTF_8)/*消息体必须是字节数组类型-->byte[]*/);System.out.println("生产者发送【 "+i+" 】条消息完成");}//5、关闭资源//关闭通道channel.close();//关闭连接conn.close();}
}

Consumer01

package cn.ljh.rabbitmq.consumer;import cn.ljh.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;//消息消费者1
public class Consumer01
{// 使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下：//（1）创建ConnectionFactory连接工厂，设置连接信息，再通过ConnectionFactory获取Connection连接。//（2）通过Connection获取Channel。//（3）根据需要、调用Channel的queueDeclare()方法声明队列，  Declare:声明、宣布//    如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列。//（4）调用Channel 的 basicConsume()方法开始处理消息，调用该方法时需要传入一个Consumer参数，该参数相当于JMS中的消息监听器。//常量public final static String QUEUE01 = "firstQueue";public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接工厂，设置连接信息，然后再通过连接工厂获取连接Connection conn = ConnectionUtil.getConnection();//2、通过Connection获取Channel 消息通道Channel channel = conn.createChannel();//3、调用 Channel 的 queueDeclare() 方法声明队列，//   如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列channel.queueDeclare(QUEUE01, /* 声明的队列名 */true,    /* 消息队列是否持久化 */false,  /* 是否只允许该消息消费者消费该队列的消息，独占 */false, /* 是否自动删除 */null   /* 指定消息队列额外的属性 */);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(QUEUE01 /*消费这个消费队列里面的消息*/,true /*消息的确认模式：是否自动确认该消息已经被消费完成并返回确认消息给消息队列*/,new DefaultConsumer(channel){//处理消息：当这个消息队列收到消息的时候，这个方法就会被触发。重写这个方法：@Overridepublic void handleDelivery(String consumerTag,Envelope envelope /*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties properties /*消息的那些属性*/,byte[] body /*body：消息的消息体*/) throws IOException{//把消息体中的消息拿出来String message = new String(body, "UTF-8");//printf：格式化输出函数   %s：输出字符串  %n：换行System.err.printf("P2PConsumer收到来自Exchange为【%s】、路由key为【%s】的消息，消息内容为%s%n",envelope.getExchange(),envelope.getRoutingKey(),message);}});}
}

Consumer02

package cn.ljh.rabbitmq.consumer;import cn.ljh.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;//消息消费者2
public class Consumer02
{// 使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下：//（1）创建ConnectionFactory连接工厂，设置连接信息，再通过ConnectionFactory获取Connection连接。//（2）通过Connection获取Channel。//（3）根据需要、调用Channel的queueDeclare()方法声明队列，  Declare:声明、宣布//    如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列。//（4）调用Channel 的 basicConsume()方法开始处理消息，调用该方法时需要传入一个Consumer参数，该参数相当于JMS中的消息监听器。//常量public final static String QUEUE02 = "secondQueue";public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接工厂，设置连接信息，然后再通过连接工厂获取连接Connection conn = ConnectionUtil.getConnection();//2、通过Connection获取Channel 消息通道Channel channel = conn.createChannel();//3、调用 Channel 的 queueDeclare() 方法声明队列，//   如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列channel.queueDeclare(QUEUE02, /* 声明的队列名 */true,    /* 消息队列是否持久化 */false,  /* 是否只允许该消息消费者消费该队列的消息，独占 */false, /* 是否自动删除 */null   /* 指定消息队列额外的属性 */);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(QUEUE02 /*消费这个名字的消费队列里面的消息*/,true/*消息的确认模式：是否自动确认该消息已经被消费完成并返回确认消息给消息队列*/,new DefaultConsumer(channel){//处理消息：当这个消息队列收到消息的时候，这个方法就会被触发。重写这个方法：@Overridepublic void handleDelivery(String consumerTag,Envelope envelope /*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties properties /*消息的那些属性*/,byte[] body /*body：消息的消息体*/) throws IOException{//把消息体中的消息拿出来String message = new String(body, "UTF-8");//printf：格式化输出函数   %s：输出字符串  %n：换行System.err.printf("P2PConsumer收到来自Exchange为【%s】、路由key为【%s】的消息，消息内容为%s%n",envelope.getExchange(),envelope.getRoutingKey(),message);}});}}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.ljh</groupId><artifactId>rabbitmq_fanout</artifactId><version>1.0.0</version><name>rabbitmq_fanout</name><!--  属性  --><properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><java.version>11</java.version></properties><!--  依赖  --><dependencies><!-- RabbitMQ 的依赖库 --><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.13.0</version></dependency></dependencies></project>

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/107066.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

[Swift]同一个工程管理多个Target

[Swift]同一个工程管理多个Target

1.准备 先创建一个测试工程“ADemo”&#xff0c;右键其Target选择Duplicate&#xff0c;再复制一个Target为“ADemo2”。 再选择TARGETS下方的“”&#xff0c;添加一个APP到项目中&#xff0c;这个命名为“BDemo”。 2、管理多个Target 可以对三个target分别导入不同的框…
阅读更多...
iWall：支持自定义的Mac动态壁纸软件

iWall：支持自定义的Mac动态壁纸软件

iWall Mac是一款动态壁纸软件&#xff0c;它可以使用任何格式的漂亮视频&#xff08;无须转换&#xff09;、图片、动画、Flash、gif、swf、程序、网页、网站做为您的动态壁纸、动态桌面&#xff0c;并且可以进行交互。 这款软件功能多、使用简单、体积小巧、不占用资源、运行…
阅读更多...
Go 1.21 新内置函数：min、max 和 clear

Go 1.21 新内置函数：min、max 和 clear

max 函数 func max[T cmp.Ordered](x T, y …T) T 这是一个泛型函数&#xff0c;用于从一组值中寻找并返回 最大值&#xff0c;该函数至少要传递一个参数。在上述函数签名中&#xff0c;T 表示类型参数&#xff0c;它必须满足 cmp.Ordered 接口中定义的数据类型要求&#xff0…
阅读更多...
多个方向工程师岗位缺人，最高25K/月！

多个方向工程师岗位缺人，最高25K/月！

高级存储工程师 岗位职责&#xff1a; 1.负责存储设备的巡检&#xff0c;及时解决设备故障&#xff1b; 2.负责定期对存储设备的运行基本状态进行健康性检查和保养等预防性维护服务&#xff1b; 3.负责提供SAN、NAS、iSCSI等存储设备的实时监控&#xff1b; 4.负责提供存储设备…
阅读更多...
微信小程序引入阿里巴巴iconfont图标并使用

微信小程序引入阿里巴巴iconfont图标并使用

介绍 在小程序里&#xff0c;使用阿里巴巴的图标&#xff0c;如下所示: 使用方式 搜索自己需要的图标&#xff0c;然后将需要用到的图标加入购物车&#xff0c;如下图所示&#xff1a; 去右上角&#xff0c;点击购物车按钮&#xff1b;这里第一次使用&#xff0c;会有三个提…
阅读更多...
Redis缓存穿透、缓存击穿、缓存雪崩详解

Redis缓存穿透、缓存击穿、缓存雪崩详解

目录 缓存处理流程 一、缓存穿透 1、概念 2、解决办法 二、缓存击穿 1、概念 2、解决办法 三、缓存雪崩 1、概念 2、解决办法 缓存处理流程 接收到查询数据请求时&#xff0c;优先从缓存中查询&#xff0c;若缓存中有数据&#xff0c;则直接返回&#xff0c;若缓存中查不到则从…
阅读更多...
再学Blazor——概述

再学Blazor——概述

简介 Blazor 是一种 .NET 前端 Web 框架&#xff0c;同时支持服务器端呈现和客户端交互性。 使用 C# 语言创建丰富的交互式 UI共享前后端应用逻辑可以生成混合桌面和移动应用受益于 .NET 的性能、可靠性和安全性需要有 HTML、CSS、JS 相关基础&#xff08;开发 UI 框架的话&a…
阅读更多...
Steam余额红锁的原因，及红锁后申诉办法

Steam余额红锁的原因，及红锁后申诉办法

安全的余额一般是通过充值卡充值获得&#xff0c;再加上交易手续费再转卖给你。一般便宜不到哪去。 但你别以为余额是安全的&#xff0c;就万事大吉了。照样有被红锁的可能性&#xff0c;比如这三种&#xff1a; 1、Steam市场巡查机制&#xff0c;红锁 平台的巡查机制和原理…
阅读更多...
在逍遥模拟器上安装LPSosed模块以及其Manager管理器

在逍遥模拟器上安装LPSosed模块以及其Manager管理器

环境&#xff1a;win7 64位&#xff0c;python3.8.10&#xff0c;逍遥模拟器9.0.6&#xff0c;安卓版本9 参考我的文章&#xff1a; 在雷电模拟器9上安装magisk并安装LSPosed模块以及其Manager管理器&#xff08;一&#xff09;-CSDN博客 前置工作&#xff1a;先开启模拟器的…
阅读更多...
一文讲解图像梯度

一文讲解图像梯度

简介&#xff1a; ​ 图像梯度计算的是图像变化的幅度。对于图像的边缘部分&#xff0c;其灰度值变化较大&#xff0c;梯度值变化也较大&#xff1b;相反&#xff0c;对于图像中比较平滑的部分&#xff0c;其灰度值变化较小&#xff0c;相应的梯度值变化也较小。一般情…
阅读更多...
测试培训机构「某峰」测评调查报告

测试培训机构「某峰」测评调查报告

测试培训机构「某峰」测评调查报告 ⭐文章简介一、机构测评之受访者介绍二、老师傅-机构测评10问&#xff01;第1问 你从哪里了解到这个培训机构&#xff1f;第2问 你为什么选择这家培训机构&#xff0c;它吸引你掏毛爷爷的点有哪些&#xff1f;第3问 回顾一下&#xff0c;从咨…
阅读更多...
CA与区块链之数字签名详解

CA与区块链之数字签名详解

CA与区块链验证本质上都是数字签名&#xff0c;首先&#xff0c;我们看一下什么是数字签名&#xff01; 数字签名 数字签名是公钥密码学中的一种技术&#xff0c;用于验证信息的完整性和发送者的身份。简而言之&#xff0c;数字签名是一种确认信息来源和信息完整性的手段。它通…
阅读更多...
ubuntu设置开机自启服务脚本

ubuntu设置开机自启服务脚本

在Ubuntu上设置开机自启服务的步骤如下&#xff1a; 创建一个服务文件&#xff1a;使用文本编辑器创建一个以 .service 结尾的服务文件&#xff0c;例如 demo-service.service。 sudo vim /etc/systemd/system/demo-service.service在服务文件中添加以下内容&#xff1a; [Unit…
阅读更多...
pandas_使用总结(1)

pandas_使用总结(1)

取值操作 常用方式 df[‘Q1’] # 选择‘Q1’列&#xff0c;同df.Q1&#xff0c;返回一个Seriesdf[[‘name’,‘Q1’]] # 选择多列&#xff0c;注意括号df[0:3] # 取前三行df[0:10:2] # 0到前10行中每两行取一次数据&#xff08;python 切片操作&#xff1a; start : end : st…
阅读更多...
企业如何选择安全又稳定的文件传输协议

企业如何选择安全又稳定的文件传输协议

企业无论是内部的数据共享&#xff0c;还是与外部的合作交流&#xff0c;都需要通过网络进行文件的传输和交换。然而&#xff0c;文件传输它涉及到多方面的因素&#xff0c;例如文件的大小、数量、类型、敏感性、传输距离、网络环境等。这些因素都会影响到文件传输的各个方面&a…
阅读更多...
“5G+北斗”赋能千行百业，中海达亮相2023中国移动全球合作伙伴大会

“5G+北斗”赋能千行百业，中海达亮相2023中国移动全球合作伙伴大会

10月12日&#xff0c;2023中国移动全球合作伙伴大会在广州保利世贸博览馆召开。本次活动以“算启新程 智享未来”为主题&#xff0c;重点展示中国移动全球合作伙伴在5G、终端、数智城市等方面的新产品、新技术和新服务。作为中国移动“朋友圈”战略合作企业及“北斗时空子链”链…
阅读更多...
快速自动化处理JavaScript渲染页面

快速自动化处理JavaScript渲染页面

在进行网络数据抓取时&#xff0c;许多网站使用了JavaScript来动态加载内容&#xff0c;这给传统的网络爬虫带来了一定的挑战。本文将介绍如何使用Selenium和ChromeDriver来实现自动化处理JavaScript渲染页面&#xff0c;并实现有效的数据抓取。 1、Selenium和ChromeDriver简介…
阅读更多...
Java中通过List中的stream流去匹配相同的字段去赋值，避免for循环去查询数据库进行赋值操作

Java中通过List中的stream流去匹配相同的字段去赋值，避免for循环去查询数据库进行赋值操作

List<EquipmentDeviceMessage> equipmentDeviceMessageInfo greenThinkTanksInfoPlanMapper.getEquipmentDeviceMessageInfo(phone, startDate, endDate); List<BladeUserVo> userList bladexsqlMapper.getUserList();Q&#xff1a;上面两个列表怎么使用流&#…
阅读更多...
Java8 新特性之Stream（三）-- Stream的终结操作

Java8 新特性之Stream（三）-- Stream的终结操作

目录 1.forEach(Consumer) 2.reduce(BinaryOperator) 3.max([Comparator]) 4.min([Comparator]) 5.count() 6.findFirst() 7.findAny() 拓展：
阅读更多...
OpenWrt如何公网ssh远程连接【内网穿透】

OpenWrt如何公网ssh远程连接【内网穿透】

文章目录 1.安装cpolar内网穿透2.配置SSH远程访问隧道3.公网远程连接4.固定远程连接地址 简单几步实现openWRT结合cpolar内网穿透工具实现远程ssh连接 1.安装cpolar内网穿透 cpolar是一个非常棒的跨平台、内网穿透工具&#xff0c;可以通过安全隧道将NAT或防火墙后面的本地服…
阅读更多...
最新文章

Copyright @ 2022~2023