RabbitMQ Topic & RPC-EW帮帮网

Topics(通配符模式)

Topics 和Routing模式的区别是:

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

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

RoutingKey 是⼀系列由点( . )分隔的单词, ⽐如 " stock.usd.nyse ", " nyse.vmw ", " quick.orange.rabbit "
BindingKey 和RoutingKey⼀样, 也是点( . )分割的字符串
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 默认值localhost
 factory.setPort(Constants.PORT); //默认值5672
 factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /
 factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guest
 factory.setPassword(Constants.PASSWORD);//密码, 默认guest
 Connection 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 默认值localhost
 factory.setPort(Constants.PORT); //默认值5672
 factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /
 factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guest
 factory.setPassword(Constants.PASSWORD);//密码, 默认guest
 Connection connection = factory.newConnection();
 Channel channel = connection.createChannel();
 //2. 接收消息, 并消费
 DefaultConsumer consumer = new DefaultConsumer(channel) {
 @Override
 public 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通信的过程, ⼤概是通过两个队列实现⼀个可回调的过程

⼤概流程如下:

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

引⼊依赖

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

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

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

声明队列

//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) {
 @Override
 public 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 默认值localhost
 factory.setPort(Constants.PORT); //默认值5672
 factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟机名称, 默认 /
 factory.setUsername(Constants.USER_NAME);//⽤⼾名,默认guest
 factory.setPassword(Constants.PASSWORD);//密码, 默认guest
 Connection 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) {
 @Override
 public 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();
 }
}

RabbitMQ Topic & RPC

Topics(通配符模式)

引⼊依赖

创建交换机

声明队列

绑定交换机和队列

发送消息

完整代码:

编写消费者代码

完整代码:

RPC(RPC通信)

引⼊依赖

声明队列

定义回调队列

使⽤内置交换机发送消息

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

获取回调结果

完整代码

网站公告

今日签到

热门文章

最新发布