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

RabbitMQ Topic RPC

news 来源：原创 2025/5/21 6:03:55

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();}
}

相关文章：

  • MS16-075 漏洞 复现过程
  • 小区服务|基于Java+vue的小区服务管理系统(源码+数据库+文档)
  • Java NIO（New I/O）
  • 【实验增效】5 μL/Test 高浓度液体试剂！Elabscience PE Anti-Mouse Ly6G抗体 简化流式细胞术流程
  • 连续空间链式推理与SoftCoT++介绍
  • 邂逅Node.js
  • IEEE 802.1Q协议下封装的VLAN数据帧格式
  • 如何管理和优化内核参数
  • CNBC专访CertiK联创顾荣辉：从形式化验证到AI赋能，持续拓展Web3.0信任边界
  • matlab编写的BM3D图像去噪方法
  • 深入浅出IIC协议 - 从总线原理到FPGA实战开发 -- 第三篇：Verilog实现I2C Master核
  • 【蓝桥杯真题精讲】第 16 届 Python A 组（省赛）
  • 网络安全之网络攻击spring临时文件利用
  • AR 开启昆虫学习新视界，解锁奇妙微观宇宙
  • 流复备机断档处理
  • 当前主流的传输技术（如OTN、IP-RAN、FlexE等）
  • 能管理MySQL、Oracle、达梦数据库的桌面管理软件开源了
  • 【神经网络与深度学习】扩散模型之原理解释
  • 第 84 场周赛：翻转图像、字符串中的查找与替换、图像重叠、树中距离之和
  • 常用UI自动化测试框架
  • 这群“工博士”，把论文“写”在车间里
  • 中国华能：1-4月新能源装机突破1亿千瓦，利润总额再创新高
  • 中国需加强自主创新和国际合作，提升产业链供应链韧性
  • 著名古人类学家高星获推选为国际史前与原史研究院院士
  • 中国古代文学研究专家、南开大学教授李剑国逝世
  • 从《缶翁的世界》看吴昌硕等湖州籍书画家对海派的影响