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;并且可以进行交互。 这款软件功能多、使用简单、体积小巧、不占用资源、运行…
阅读更多...
微信小程序引入阿里巴巴iconfont图标并使用

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

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

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

目录 缓存处理流程 一、缓存穿透 1、概念 2、解决办法 二、缓存击穿 1、概念 2、解决办法 三、缓存雪崩 1、概念 2、解决办法 缓存处理流程 接收到查询数据请求时&#xff0c;优先从缓存中查询&#xff0c;若缓存中有数据&#xff0c;则直接返回&#xff0c;若缓存中查不到则从…
阅读更多...
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;数字签名是一种确认信息来源和信息完整性的手段。它通…
阅读更多...
企业如何选择安全又稳定的文件传输协议

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

企业无论是内部的数据共享&#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简介…
阅读更多...
OpenWrt如何公网ssh远程连接【内网穿透】

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

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

【Springboot】Filter 过滤器的使用

一、基本介绍 过滤器 Filter 作为 Java 三大器之一&#xff0c;在 Java Web 的使用中有很高的地位。所谓过滤器&#xff0c;就是实现了 javax.servlet.Filter 接口的服务器端程序&#xff0c;就是对事物进行过滤的。在 Web 中的过滤器&#xff0c;当然就是对请求进行过滤&#…
阅读更多...
解决axios不发起请求的问题

解决axios不发起请求的问题

一个很简单axios发起的get请求&#xff0c;但是network就是没有xhr记录&#xff1a; 找了半天也没有找到原因&#xff0c;后来问了chatgpt才找到原因&#xff1a; 请求地址url前面没有const关键字&#xff0c;此时url相当于undefined&#xff0c;而axios在url为undefined时不会…
阅读更多...
天猫商品品牌数据采集接口，天猫商品详情数据接口，天猫API接口

天猫商品品牌数据采集接口，天猫商品详情数据接口，天猫API接口

天猫商品品牌数据采集方法如下&#xff1a; 打开天猫&#xff0c;进入任意一个品牌的商品页面。点击浏览器右上角的“选项”按钮&#xff0c;选择“检查元素”或使用快捷键CtrlShiftI&#xff08;Windows&#xff09;或CmdOptionI&#xff08;Mac&#xff09;打开开发者工具。…
阅读更多...
JVM类装载器详解

JVM类装载器详解

目录 一、类装载的过程 1.1 装载(Load) 1.2 链接(Link) 1.2.1 验证(Varify) 二、类装载器组成 1. JVM 中内置了三个重要的 ClassLoader&#xff0c;同时按如下顺序进行加载&#xff1a; 2、图解 3、加载原则 所谓的双亲委派 类加载器负责在运行时将Java类动态加载到Java虚拟机&…
阅读更多...
多关键词高亮显示

多关键词高亮显示

引入关键词文件&#xff0c;符合有条件的背景色高亮显示&#xff0c;也可取消。 <div id"testHtml"><p>写入的文本</p><p>关键词</p></div> var str 多个关键词&#xff0c;关键词文件&#xff0c;关键词 var strL str.replac…
阅读更多...
nginx报错

nginx报错

故障1&#xff1a;nginx 502错误 故障描述 一个 post 的请求&#xff0c;直接调接口服务数据正常返回&#xff0c;但是通过 nginx 代理后&#xff0c; 什么都没有返回 nginx错误日志 upstream sent invalid chunked response while reading upstream 原因 http协议版本不一致…
阅读更多...
【C++】C++11—— 包装器

【C++】C++11—— 包装器

​ ​&#x1f4dd;个人主页&#xff1a;Sherry的成长之路 &#x1f3e0;学习社区&#xff1a;Sherry的成长之路&#xff08;个人社区&#xff09; &#x1f4d6;专栏链接&#xff1a;C学习 &#x1f3af;长路漫漫浩浩&#xff0c;万事皆有期待 上一篇博客&#xff1a;【C】C11…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023