203、RabbitMQ 之 使用 direct 类型的 Exchange 实现 消息路由 (RoutingKey)

目录

  • ★ 使用direct实现消息路由
  • 代码演示这个情况二
    • ConstantUtil 常量工具类
    • ConnectionUtil 连接RabbitMQ的工具类
    • Publisher 消息生产者
      • 测试消息生产者
    • Consumer01 消息消费者01
      • 测试消费者结果:
    • Consumer02 消息消费者02
      • 测试消费者结果:
  • 完整代码:
    • ConstantUtil
    • ConnectionUtil
    • Publisher
    • Consumer01
    • Consumer02
    • pom.xml

★ 使用direct实现消息路由

direct类型的Exchange: 生产者发送给Exchange 消息的路由key 要和 Exchange 绑定 Queue 消息队列的路由key一致,这个生产者发送的消息才会被Exchange 分发到那个消息队列里面去。

在这里插入图片描述

direct类型的Exchange会根据消息的路由key将消息分发给指定的Queue。

▲ 一个队列能与一个Exchange绑定多个路由key,
Q1队列与Exchange就绑定的路由key:orange
比如Q2队列与Exchange就绑定了两个路由key:black和green。

若消息生产者发送路由key为orange的消息到Exchange时,该消息将会被分发到Q1队列;

若发送路由key为black或green的消息到Exchange时,该消息都将被分发到Q2队列。

▲ 情况一

RabbitMQ也允许多个队列绑定相同的路由key,此时又变成了Pub-Sub模型。

比如C1、C2两个队列都绑定了black作为路由key,这意味着若消息生产者发送路由key为black的消息时,该消息将会被分发Q1和Q2两个队列,Q1、Q2两个队列将会收到各自不同副本——这就是Pub-Sub模型。
在这里插入图片描述

▲ 情况二

C1队列绑定了error作为路由key,而Q2则绑定了info、error和warning作为路由key,

这意味着若消息生产者发送路由key为error的消息时,该消息将会被分发Q1和Q2两个队列,Q1、Q2两个队列将会同时收到各自的副本;

若消息生产者发送路由key为info或warning的消息时,该消息只会被分发给Q2队列,Q1队列不会收到任何消息。

在这里插入图片描述

代码演示这个情况二

需求:就是演示这张图
在这里插入图片描述
过程:
1、创建一个消息生产者,2个消息消费者,
2、生产者声明一个Exchange,2个消息队列,
然后如图,Exchange绑定Q1这个消息队列用1个路由key,Exchange绑定Q2这个消息队列用3个路由key,
3、生产者发送30条消息给Exchange,每个路由key发送10条。
结果应该是:Q1消息队列得到10条消息,Q2消息队列得到30条消息。
4、创建2个消费者,分别消费Q1和Q2.

ConstantUtil 常量工具类

在这里插入图片描述

ConnectionUtil 连接RabbitMQ的工具类

在这里插入图片描述

Publisher 消息生产者

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

在这里插入图片描述

测试消息生产者

如图:结果正确。
消息队列queue_01,因为只绑定一个路由key–error,所以exchange分发10条路由key为error的消息给它。
消息队列queue_01,因为只绑定3个路由key–error、info、warning,所以exchange分发30条消息给它。
在这里插入图片描述

在这里插入图片描述

Consumer01 消息消费者01

消息消费者就没啥好说的,
Consumer01 消费 queue_01 消息队列,Consumer02 消费 queue_02 消息队列。
代码差不多。

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

测试消费者结果:

正常成功,符合期望
在这里插入图片描述

Consumer02 消息消费者02

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

测试消费者结果:

正常成功,符合期望
在这里插入图片描述




完整代码:

ConstantUtil

package cn.ljh.rabbitmq.util;//常量
public class ConstantUtil
{// 消息队列的名称public final static String QUEUE01 = "queue_01";public final static String QUEUE02 = "queue_02";// Exchange的名称public static final String EXCHANGE_NAME = "myex02.direct";// 三个路由key定义成一个数组的名称public static final String[] ROUTING_KEYS = {"info", "error", "warning"};}

ConnectionUtil

package cn.ljh.rabbitmq.util;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
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 cn.ljh.rabbitmq.util.ConstantUtil;
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;//消息生产者--使用 direct 类型的exchange------就是广播模式
public class Publisher
{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(ConstantUtil.EXCHANGE_NAME,/* Exchange名字 */BuiltinExchangeType.DIRECT,/* Exchange 类型 */true,/* 是否持久化 */false,/* 是否自动栅除 */false,/* 是否为内部的 Exchange */null /* 指定 Exchange 的额外属性 */);//调用queueDeclare()方法声明队列----声明多个消息队列------声明第1个消息队列---------路由key是 errorchannel.queueDeclare(ConstantUtil.QUEUE01, true, false, false, null);//把 Exchange 和 Queue 绑定起来,绑定第一个消息队列channel.queueBind(ConstantUtil.QUEUE01, ConstantUtil.EXCHANGE_NAME,ConstantUtil.ROUTING_KEYS[1] /* exchange类型是direct,这里指定路由key */,null /* 指定 Exchange 的额外属性 */);//声明第2个消息队列--------这个exchange绑定这个queue,使用了三个路由key---info,error,warningchannel.queueDeclare(ConstantUtil.QUEUE02, true, false, false, null);//用循环为第2个消息队列绑定三个路由keyfor (int i = 0; i < ConstantUtil.ROUTING_KEYS.length; i++){//把 Exchange 和 Queue 绑定起来,绑定第2个消息队列channel.queueBind(ConstantUtil.QUEUE02,ConstantUtil.EXCHANGE_NAME,ConstantUtil.ROUTING_KEYS[i]  /* 循环绑定路由key */,null  /* 指定 Exchange 的额外属性 */);}//生产者发送30条消息for (int i = 1; i <= 30; i++){//用嵌套的三元运算符,动态的让消息的发送使用到不同的路由keyString routingKey = i <= 10 ? ConstantUtil.ROUTING_KEYS[0] :(i <= 20 ? ConstantUtil.ROUTING_KEYS[1] : ConstantUtil.ROUTING_KEYS[2]);//要发送的消息String message = "生产者发送的第【 " + i + " 】条消息的内容";//4、调用Channel 的 basicPublish() 方法发送消息channel.basicPublish(ConstantUtil.EXCHANGE_NAME /* 指定向这个Exchange发送消息 */,routingKey /* 动态指定路由key */,null /*指定额外的消息的属性*/,message.getBytes(StandardCharsets.UTF_8)/*消息体必须是字节数组类型-->byte[]*/);System.out.println("生产者发送【 " + i + " 】条消息完成,发送到Exchange的路由key是:"+routingKey);}//5、关闭资源//关闭通道channel.close();//关闭连接conn.close();}
}

Consumer01

package cn.ljh.rabbitmq.consumer;import cn.ljh.rabbitmq.util.ConnectionUtil;
import cn.ljh.rabbitmq.util.ConstantUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;
// 使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下:
//(1)创建ConnectionFactory连接工厂,设置连接信息,再通过ConnectionFactory获取Connection连接。
//(2)通过Connection获取Channel。
//(3)根据需要、调用Channel的queueDeclare()方法声明队列,  Declare:声明、宣布
//    如果声明的队列已存在,该方法直接获取已有的队列;如果声明的队列还不存在,该方法将会创建新的队列。
//(4)调用Channel 的 basicConsume()方法开始处理消息,调用该方法时需要传入一个Consumer参数,该参数相当于JMS中的消息监听器。//消息消费者1
public class Consumer01
{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(ConstantUtil.QUEUE01, /* 声明的队列名 */true,    /* 消息队列是否持久化 */false,  /* 是否只允许该消息消费者消费该队列的消息,独占 */false, /* 是否自动删除 */null   /* 指定消息队列额外的属性 */);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(ConstantUtil.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 cn.ljh.rabbitmq.util.ConstantUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;//消息消费者2
public class Consumer02
{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(ConstantUtil.QUEUE02, /* 声明的队列名 */true,    /* 消息队列是否持久化 */false,  /* 是否只允许该消息消费者消费该队列的消息,独占 */false, /* 是否自动删除 */null   /* 指定消息队列额外的属性 */);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(ConstantUtil.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>direct</artifactId><version>1.0.0</version><name>direct</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><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.16</version><scope>compile</scope></dependency></dependencies></project>

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

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

相关文章

机器学习(22)---信息熵、纯度、条件熵、信息增益

文章目录 1、信息熵2、信息增益3、例题分析 1、信息熵 1. 信息熵(information entropy)是度量样本集合纯度最常用的一种指标。信息的混乱程度越大&#xff0c;不确定性越大&#xff0c;信息熵越大&#xff1b;对于纯度&#xff0c;就是信息熵越大&#xff0c;纯度越低。 2. 纯度…

Text-to-SQL小白入门(八)RLAIF论文:AI代替人类反馈的强化学习

学习RLAIF论文前&#xff0c;可以先学习一下基于人类反馈的强化学习RLHF&#xff0c;相关的微调方法&#xff08;比如强化学习系列RLHF、RRHF、RLTF、RRTF&#xff09;的论文、数据集、代码等汇总都可以参考GitHub项目&#xff1a;GitHub - eosphoros-ai/Awesome-Text2SQL: Cur…

论文阅读/写作扫盲

第一节&#xff1a;期刊科普 JCR分区和中科院分区是用于对期刊进行分类和评估的两种常见方法。它们的存在是为了帮助学术界和研究人员更好地了解期刊的学术质量、影响力和地位。 JCR分区&#xff08;Journal Citation Reports&#xff09;&#xff1a;JCR分区是由Clarivate Ana…

android U广播详解(一)

概念介绍 进程队列 BroadcastQueueModernImpl 的设计围绕着为设备上的每个潜在进程维护一个单独的 BroadcastProcessQueue 实例。表明用于传送到特定进程的Pending {link BroadcastRecord} 条目队列。整个类都标记为 {code NotThreadSafe}&#xff0c;因为调用者有责任始终与…

Excel往Word复制表格时删除空格

1.背景 在编写文档&#xff0c;经常需要从Excel往Word里复制表格 但是复制过去的表格前面会出现空格&#xff08;缩进&#xff09; 再WPS中试了很多方法&#xff0c;终于摆脱了挨个删除的困扰 2. WPS排版中删除 选择表格菜单栏-选在【开始】-【排版】选择【更多段落处理】-【段…

ROS-PX4仿真笔记_1

offbord模式测试 rosrun offboard_pkg position stablelize模式 lqr控制器实验 roslaunch px4 fast_test.launch 无人机起飞1.5-2m sh mybot_gazebo.sh#roslaunch px4 fast_racing.launch & sleep 20; roslaunch ego_planner single_run_in_gazebo.launch & sleep 1…

快速解决“找不到msvcr120.dll无法执行代码”问题,总结5解决方法

计算机已经成为我们生活和工作中不可或缺的一部分。然而&#xff0c;在使用计算机的过程中&#xff0c;我们常常会遇到各种问题&#xff0c;其中之一就是找不到msvcr120.dll文件。这个问题可能会可能导致计算机程序软件&#xff0c;游戏无法正常运行&#xff0c;影响到我们的工…

使用testMe自动生成单元测试用例

文章目录 1、testMe简介2、插件对比2.1 testMe2.2 Squaretest2.3 Diffblue 3、IDEA插件安装4、单测用例4.1 maven依赖4.2 生成用例 5、自定义模板6、使用自定义模板生成用例7、调试用例 1、testMe简介 公司对于系统单元测试覆盖率有要求&#xff0c;需要达到50%或80%以上才可以…

身份证号码,格式校验:@IdCard(自定义注解)

目标 自定义一个用于校验 身份证号码 格式的注解IdCard&#xff0c;能够和现有的 Validation 兼容&#xff0c;使用方式和其他校验注解保持一致&#xff08;使用 Valid 注解接口参数&#xff09;。 校验逻辑 有效格式 符合国家标准。 公民身份号码按照GB11643&#xff0d;…

使用 Eziriz .NET Reactor 对c#程序加密

我目前测试过好几个c#加密软件。效果很多时候是加密后程序执行错误&#xff0c;或者字段找不到的现象 遇到这个加密软件用了一段时间都很正常&#xff0c;分享一下使用流程 破解版本自行百度。有钱的支持正版&#xff0c;我用的是 Eziriz .NET Reactor 6.8.0 第一步 安装 Ezi…

我用了多年的前端框架,强烈推荐!

大家好&#xff0c;我是鱼皮&#xff0c;今天给大家分享一个我自己用了多年、现在团队也在用的前端框架 —— Ant Design Pro。 什么是 Ant Design Pro&#xff1f; Ant Design Pro 是由蚂蚁金服开发的、基于 Ant Design 组件库的开发框架&#xff0c;专门用于构建企业级管理…

《UnityShader入门精要》学习1

读者可以在开源网站github&#xff08;https://github.com/candycat1992/Unity_Shaders_Book&#xff09;上下载本书的源代码。 第二章 渲染流水线 渲染流水线的最终目的在于生成或者说是渲染一张二维纹理&#xff0c;即我们在电脑屏幕上看到的所有效果&#xff0c;它的输入是…