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.rhkb.cn/news/157355.html

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

相关文章

代理和多级代理

文章目录 代理使用场景代理过程实验演示多级代理 代理使用场景 1、拿下远程 web 服务器 2、webshell 链接不稳定&#xff0c;需要使用稳定的木马程序 3、远程服务器无法直接链接攻击者电脑 4、需要借助公网vps转发来自失陷服务器的木马流量 5、借助frp服务端(vps)和客户端(内网…

实现Element Select选择器滚动加载

<template><el-selectpopper-class"more-tag-data"v-model"tagId"filterableplaceholder"请选择"focus"focusTag"><el-optionv-for"(item, index) in taskTagLists":key"index":label"item.n…

如何在小程序中设置导航栏文字颜色和背景颜色

不同商家有不同的颜色风格&#xff0c;例如有些做设计的公司&#xff0c;主要是黑色风格&#xff1b;有些卖珠宝的商家&#xff0c;主要是金色风格&#xff1b;他们的小程序&#xff0c;也需要进行同样的风格设定。下面具体介绍怎么在小程序中进行整个风格设定。 1. 在小程序管…

破局「二次创业」:合思的新解法

在新的水温下&#xff0c;寻找更为良性的发展正在成为企业的必答题。对此&#xff0c;合思给出的不仅是一份更“省”的答题方法。也更是从认知层到行动层&#xff0c;最后到工具层的一张授人以渔的“渔网”。 作者|思杭 编辑|皮爷 出品|产业家 今年4月初&#xff0c;广州…

lvgl 界面管理器

lv_scr_mgr lvgl 界面管理器 适配 lvgl 8.3 降低界面之间的耦合使用较小的内存&#xff0c;界面切换后会自动释放内存内存泄漏检测 使用方法 在lv_scr_mgr_port.h 中创建一个枚举&#xff0c;用于界面ID为每个界面创建一个页面管理器句柄将界面句柄添加到 lv_scr_mgr_por…

【基础篇】三、Flink集群角色、系统架构以及作业提交流程

文章目录 1、集群角色2、部署模式3、Flink系统架构3.1 作业管理器&#xff08;JobManager&#xff09;3.2 任务管理器&#xff08;TaskManager&#xff09; 4、独立部署会话模式下的作业提交流程5、Yarn部署的应用模式下作业提交流程 1、集群角色 Flink提交作业和执行任务&…

EXCEL——根据单元格值设置不同色阶

方法&#xff1a;开始—>条件格式—>色阶&#xff08;默认色阶或复杂色阶&#xff09;。 一、默认色阶 如图&#xff0c;可选择自定义的色阶模式。 二、复杂色阶 1、如图&#xff0c;点击"其他规则" 2、选择复杂格式 此时可以看到&#xff0c;支持多种格式…

有 AI,无障碍,AIoT 设备为视障人群提供便利

据世界卫生组织统计&#xff0c;全球共 22 亿人视力受损&#xff0c;包含 2.85 亿视障人群和 3,900 万全盲人群。而且&#xff0c;这一数字将随老龄化加剧不断增加。 虽然视障人群面临着诸多不便&#xff0c;但是针对视障人群的辅助设备却存在成本高、维护困难、操作复杂等问题…

给你一个项目,你将如何开展性能测试工作?

一、性能三连问 1、何时进行性能测试&#xff1f; 性能测试的工作是基于系统功能已经完备或者已经趋于完备之上的&#xff0c;在功能还不够完备的情况下没有多大的意义。因为后期功能完善上会对系统的性能有影响&#xff0c;过早进入性能测试会出现测试结果不准确、浪费测试资…

MQTT C库下载

方法一、从Eclipse paho下载 https://eclipse.dev/paho/index.php?pagedownloads.php 方法二&#xff0c;从MQTT官网下载 https://mqtt.org/software/ https://os.mbed.com/teams/mqtt/code/MQTTPacket/ MQTTPacket源码和paho下载的差不多 方法三、从Keil5 包管理工具…

centos7安装db2 version11.1

centos7安装DB2 操作系统 linux centos7 DB2版本 11.1 1、取包 IBM MRS Tool 将安装包放在 /home/software 下面 mkdir -p /home/software cd /home/software wget https://iwm.dhe.ibm.com/sdfdl/v2/regs2/db2pmopn/Express-C/DB2ExpressC11/Xa.2/Xb.aA_60_-i7wWKFMFpbW1xl1…

【数据结构】什么是线性表?

&#x1f984;个人主页:修修修也 &#x1f38f;所属专栏:数据结构 ⚙️操作环境:Visual Studio 2022 目录 线性表的定义 生活中常见的线性表 结语 线性表的定义 线性结构的特点: 在数据元素的非空有限集中: 存在唯一的一个被称作"第一个"的数据元素;存在唯一的一个…

10个打工人必备AI神器,升职加薪靠AI

HI&#xff0c;同学们&#xff0c;我是赤辰&#xff0c;本期是第18篇AI工具类教程&#xff0c;文章底部准备了粉丝福利&#xff0c;看完后可领取&#xff01;1. Runway&#xff08;文字转视频AI工具&#xff09; 只需要一句提示词就能精确生成你所想象的视频场景&#xff0c;还…

nnunetv2训练报错 ValueError: mmap length is greater than file size

目录 报错解决办法 报错 笔者在使用 nnunetv2 进行 KiTS19肾脏肿瘤分割实验的训练步骤中 使用 2d 和3d_lowres 训练都没有问题 nnUNetv2_train 40 2d 0nnUNetv2_train 40 3d_lowres 0但是使用 3d_cascade_fullres 和 3d_fullres 训练 nnUNetv2_train 40 3d_cascade_fullres …

Prometheus:优秀和强大的监控报警工具

文章目录 概述Prometheus的底层技术和原理数据模型数据采集数据存储查询语言数据可视化 Prometheus的部署Prometheus的使用配置数据采集目标查询监控数据设置警报规则 查看数据可视化总结 概述 Prometheus是一款开源的监控和警报工具&#xff0c;用于收集和存储系统和应用程序…

Xcode 14.3.1build 报错整理

1、Command PhaseScriptExecution failed with a nonzero exit code 2、In /Users/XX/XX/XX/fayuan-mediator-app-rn/ios/Pods/CocoaLibEvent/lib/libevent.a(buffer.o), building for iOS Simulator, but linking in object file built for iOS, file /Users/XX/XX/XX/fayuan…

软件设计之抽象工厂模式

抽象工厂模式指把一个产品变成一个接口&#xff0c;它的子产品作为接口的实现&#xff0c;所以还需要一个总抽象工厂和它的分抽象工厂。 下面我们用一个案例去说明抽象工厂模式。 在class中可以选择super类和medium类&#xff0c;即选择一个产品的子类。在type中可以选择产品的…

OJ项目——统一数据格式返回,我是如何处理的?

目录 前言 OJ项目中是如何处理的 1、准备一个类&#xff0c;作为统一的数据返回格式 2、准备一个类&#xff0c;实现ResponseBodyAdvice接口 3、我们如何写返回值更好 4、进一步优化返回值 小结 前言 关于SpringBoot的同一功能处理&#xff0c;本博主在这篇博客已经有介…

竞赛 深度学习 机器视觉 车位识别车道线检测 - python opencv

0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; 深度学习 机器视觉 车位识别车道线检测 该项目较为新颖&#xff0c;适合作为竞赛课题方向&#xff0c;学长非常推荐&#xff01; &#x1f947;学长这里给一个题目综合评分(每项满分5分) …

学习编程-先改变心态

编程失败的天才 林一和我很久以前就认识了——我从五年级就认识他了。他是班上最聪明的孩子。如果每个人在家庭作业或考试准备方面需要帮助&#xff0c;他们都会去那里。 有趣的是&#xff0c;林一不是那种连续学习几个小时的孩子。 他的聪明才智似乎与生俱来&#xff0c;几乎毫…