当前位置： 首页 > java >正文

RabbitMQ Topic RPC

java 2025/8/21 17:34:47

Topics(通配符模式)

Topics 和Routing模式的区别是:

  1. topics 模式使⽤的交换机类型为topic(Routing模式使⽤的交换机类型为direct)
  2. topic 类型的交换机在匹配规则上进⾏了扩展, Binding Key⽀持通配符匹配(direct类型的交换机路 由规则是BindingKey和RoutingKey完全匹配)

在topic类型的交换机在匹配规则上, 有些要求:

  1. RoutingKey 是⼀系列由点( . )分隔的单词, ⽐如 " stock.usd.nyse ", " nyse.vmw ", " quick.orange.rabbit "
  2. BindingKey 和RoutingKey⼀样, 也是点( . )分割的字符串
  3. Binding Key中可以存在两种特殊字符串, ⽤于模糊匹配
  • * 表⽰⼀个单词
  • # 表⽰多个单词(0-N个)

⽐如:

  • Binding Key 为"d.a.b" 会同时路由到Q1 和Q2
  • Binding Key 为"d.a.f" 会路由到Q1
  • Binding Key 为"c.e.f" 会路由到Q2
  • Binding Key 为"d.b.f" 会被丢弃, 或者返回给⽣产者(需要设置mandatory参数)

引⼊依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version>
</dependency>

编写⽣产者代码 和路由模式, 发布订阅模式的区别是:

交换机类型不同, 绑定队列的RoutingKey不同

创建交换机

定义交换机类型为BuiltinExchangeType.TOPIC

channel.exchangeDeclare(Constants.TOPIC_EXCHANGE_NAME,
BuiltinExchangeType.TOPIC, true, false, false, null);

声明队列

channel.queueDeclare(Constants.TOPIC_QUEUE_NAME1, true, false, false, null);
channel.queueDeclare(Constants.TOPIC_QUEUE_NAME2, true, false, false, null);

绑定交换机和队列

//队列1绑定error, 仅接收error信息
channel.queueBind(Constants.TOPIC_QUEUE_NAME1,Constants.TOPIC_EXCHANGE_NAME,
"*.error");
//队列2绑定info, error: error,info信息都接收
channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"#.info");
channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"*.error");

发送消息

String msg = "hello topic, I'm order.error";
channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.error",null,msg.getBy
tes());
String msg_black = "hello topic, I'm order.pay.info";
channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.pay.info",null,msg_bl
ack.getBytes());
String msg_green= "hello topic, I'm pay.error";
channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"pay.error",null,msg_green.g
etBytes());

完整代码:

public static String TOPIC_EXCHANGE_NAME = "test_topic";
public static String TOPIC_QUEUE_NAME1 = "topic_queue1";
public static String TOPIC_QUEUE_NAME2 = "topic_queue2";import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import constant.Constants;
public class TopicRabbitProducer {public static void main(String[] args) throws Exception {//1. 创建channel通道ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);//ip 默认值localhostfactory.setPort(Constants.PORT); //默认值5672factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guestfactory.setPassword(Constants.PASSWORD);//密码, 默认guestConnection connection = factory.newConnection();Channel channel = connection.createChannel();//2. 创建交换机channel.exchangeDeclare(Constants.TOPIC_EXCHANGE_NAME,
BuiltinExchangeType.TOPIC, true, false, false, null);//3. 声明队列//如果没有⼀个这样的⼀个队列, 会⾃动创建, 如果有, 则不创建channel.queueDeclare(Constants.TOPIC_QUEUE_NAME1, true, false, false,
null);channel.queueDeclare(Constants.TOPIC_QUEUE_NAME2, true, false, false,
null);//4. 绑定队列和交换机//队列1绑定error, 仅接收error信息channel.queueBind(Constants.TOPIC_QUEUE_NAME1,Constants.TOPIC_EXCHANGE_NAME,
"*.error");//队列2绑定info, error: error,info信息都接收channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"#.info");channel.queueBind(Constants.TOPIC_QUEUE_NAME2,Constants.TOPIC_EXCHANGE_NAME,
"*.error");//5. 发送消息String msg = "hello topic, I'm order.error";channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.error",null,msg.getBy
tes());String msg_black = "hello topic, I'm order.pay.info";channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"order.pay.info",null,msg_bl
ack.getBytes());String msg_green= "hello topic, I'm pay.error";channel.basicPublish(Constants.TOPIC_EXCHANGE_NAME,"pay.error",null,msg_green.g
etBytes());//6.释放资源channel.close();connection.close();}
}

编写消费者代码

Routing模式的消费者代码和Routing模式代码⼀样, 修改消费的队列名称即可

同样复制出来两份

消费者1:TopicRabbitmqConsumer1

消费者2: TopicRabbitmqConsumer2

完整代码:

import com.rabbitmq.client.*;
import constant.Constants;
import java.io.IOException;
public class TopicRabbitmqConsumer1 {public static void main(String[] args) throws Exception {//1. 创建channel通道ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);//ip 默认值localhostfactory.setPort(Constants.PORT); //默认值5672factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guestfactory.setPassword(Constants.PASSWORD);//密码, 默认guestConnection connection = factory.newConnection();Channel channel = connection.createChannel();//2. 接收消息, 并消费DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息: " + new String(body));}};channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, true, consumer);}
}

运⾏程序, 观察结果

运⾏⽣产者, 可以看到队列的消息数

运⾏消费者

RPC(RPC通信)

RPC(Remote Procedure Call), 即远程过程调⽤. 它是⼀种通过⽹络从远程计算机上请求服务, ⽽不需要 了解底层⽹络的技术. 类似于Http远程调⽤

RabbitMQ实现RPC通信的过程, ⼤概是通过两个队列实现⼀个可回调的过程

⼤概流程如下:

  1. 客⼾端发送消息到⼀个指定的队列, 并在消息属性中设置replyTo字段, 这个字段指定了⼀个回调队 列, 服务端处理后, 会把响应结果发送到这个队列
  2. 服务端接收到请求后, 处理请求并发送响应消息到replyTo指定的回调队列
  3. 客⼾端在回调队列上等待响应消息. ⼀旦收到响应，客⼾端会检查消息的correlationId属性，以确 保它是所期望的响应

引⼊依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version>
</dependency>

编写客⼾端代码 客⼾端代码主要流程如下:

  1. 声明两个队列, 包含回调队列replyQueueName, 声明本次请求的唯⼀标志corrId
  2. 将replyQueueName和corrId配置到要发送的消息队列中
  3. 使⽤阻塞队列来阻塞当前进程, 监听回调队列中的消息, 把请求放到阻塞队列中
  4. 阻塞队列有消息后, 主线程被唤醒，打印返回内容

声明队列

//2. 声明队列, 发送消息
channel.queueDeclare(Constants.RPC_REQUEST_QUEUE_NAME, true, false, false,
null);

定义回调队列

// 定义临时队列，并返回⽣成的队列名称
String replyQueueName = channel.queueDeclare().getQueue();

使⽤内置交换机发送消息

// 本次请求唯⼀标志
String corrId = UUID.randomUUID().toString();
// ⽣成发送消息的属性
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId) // 唯⼀标志本次请求.replyTo(replyQueueName) // 设置回调队列.build();
// 通过内置交换机, 发送消息
String message = "hello rpc...";
channel.basicPublish("", Constants.RPC_REQUEST_QUEUE_NAME, props,
message.getBytes());

使⽤阻塞队列, 来存储回调结果

// 阻塞队列，⽤于存储回调结果
final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
//接收服务端的响应
DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到回调消息:"+ new String(body));//如果唯⼀标识正确, 放到阻塞队列中if (properties.getCorrelationId().equals(corrId)) {response.offer(new String(body, "UTF-8"));}}
};
channel.basicConsume(replyQueueName, true, consumer);

获取回调结果

// 获取回调的结果
String result = response.take();
System.out.println(" [RPCClient] Result:" + result);

完整代码

public static String RPC_REQUEST_QUEUE_NAME = "rpc_request_queue";import com.rabbitmq.client.*;
import constant.Constants;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class RPCClient {public static void main(String[] args) throws Exception {//1. 创建Channel通道ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);//ip 默认值localhostfactory.setPort(Constants.PORT); //默认值5672factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guestfactory.setPassword(Constants.PASSWORD);//密码, 默认guestConnection connection = factory.newConnection();Channel channel = connection.createChannel();//2. 声明队列channel.queueDeclare(Constants.RPC_REQUEST_QUEUE_NAME, true, false,
false, null);// 唯⼀标志本次请求String corrId = UUID.randomUUID().toString();// 定义临时队列，并返回⽣成的队列名称String replyQueueName = channel.queueDeclare().getQueue();// ⽣成发送消息的属性AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId) // 唯⼀标志本次请求.replyTo(replyQueueName) // 设置回调队列.build();// 通过内置交换机, 发送消息String message = "hello rpc...";channel.basicPublish("", Constants.RPC_REQUEST_QUEUE_NAME, props,
message.getBytes());// 阻塞队列，⽤于存储回调结果final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);//接收服务端的响应DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到回调消息:"+ new String(body));if (properties.getCorrelationId().equals(corrId)) {response.offer(new String(body, "UTF-8"));}}};channel.basicConsume(replyQueueName, true, consumer);// 获取回调的结果String result = response.take();System.out.println(" [RPCClient] Result:" + result);//释放资源channel.close();connection.close();}
}

http://www.xdnf.cn/news/7460.html

相关文章：

  • 在Windows 11中，Edge浏览器默认会打开多个标签页，导致任务切换时标签页过多
  • List更简洁的编码构建
  • 【华为鸿蒙电脑】首款鸿蒙电脑发布：MateBook Fold 非凡大师 MateBook Pro，擎云星河计划启动
  • 易趋赋能智能家电：从需求到交付的全链路降本增效
  • 【Jitsi Meet】(腾讯会议的平替)Docker安装Jitsi Meet指南-使用内网IP访问
  • 聚焦开放智能，抢占技术高地 | 2025 高通边缘智能创新应用大赛第五场公开课来袭！
  • ⼆叉搜索树详解
  • 《MambaLLIE：基于隐式Retinex感知的低光照增强框架与全局-局部状态空间建模》学习笔记
  • 测试--自动化测试函数
  • C++类与对象--4 友元
  • 【C++】日期类
  • sherpa-ncnn：音频处理跟不上采集速度 -- 语音转文本大模型
  • Logrotate：配置日志轮转、高效管理Linux日志文件
  • 开发体育比分网站，有哪些坑需要注意的
  • 手搓一个Transformer
  • 以用户为中心的产品才是好产品
  • Kali安装配置JAVA环境和切换JDK版本的最详细的过程
  • BGP综合实验（2）
  • ai agent(智能体)开发 python高级应用7: crawl4ai 0.6.3 加re正则表达式 获取百度中含有 韩立的图片要求横屏图片
  • ts导入vue文件时提示找不到模块或其相应的类型声明问题解决
  • ADVANTEST Q8326光学波长计操作手Operation Manual
  • 升级mysql (rpm安装)
  • MIMO 检测(6)--最大似然检测(1)
  • js逆向反调试的基本 bypass
  • 【C语言】大程序结构
  • Linux详解基本指令（一）
  • 对盒模型的理解
  • 澳大利亚TikTok网络专线+本地化策略：澳洲电商品牌的破局之道
  • 最大子树和--树形dp
  • day30python打卡