本节继续探讨一些RabbitMQ的高阶用法以及应用场景。并结合之前的实例给出RabbitMQ的消息传输保障层级。
1. RabbitMQ 实现RPC调用
RPC即Remote Procedure Call，即远程过程调用。概念可参考 远程过程调用协议。
PRC协议有很多，比如CORBA、Java RMI、Thrift、Restful API等。
我试着总结了下RabbitMQ实现RPC的几个特点:
- 实现异步的远程调用
- 可以暂存请求,缓解服务端压力
-
编程复杂,并且由于加入了新的节点,可能出现重复调用的问题.
其实质可以看做是多生产者-多消费者-双队列的一个模型,在集群环境下有一定的应用场景.
图片
根据上图所示,RPC的处理流程如下:
- 客户端启动时,创建一个匿名的回调队列
- 客户端为PRC请求设置两个属性:replyTo告之RPC服务端恢复请求时的回调队列, correlationId用来标记关联id
- 请求发送到rpc_queue中
- 服务端监听rpc_queue,处理请求,并将结果发送给reply_to设定的回调队列.
- 客户端监听回调队列.检查关联id correlationId.
下面是一个简单的服务端和客户端demo:
- 服务端:
package com.pctf.advance;
import com.pctf.basic.ConnectionCreator;
import com.rabbitmq.client.*;
import java.io.IOException;
public class RPCServer {
private static final String RPC_QUEUE_NAME = "rpc_queue";
private static final String RPC_EXCHANGE_NAME = "rpc_exchange";
private static final String RPC_ROUTING_NAME = "rpc_routing";
public static void main(String[] args) throws Exception {
new RPCServer().startServer();
}
public void startServer() throws IOException {
Connection connection = ConnectionCreator.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(RPC_QUEUE_NAME, true, false, false, null);
channel.exchangeDeclare(RPC_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
channel.queueBind(RPC_QUEUE_NAME, RPC_EXCHANGE_NAME, RPC_ROUTING_NAME);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// 消息参数中取出结果的回调队列
String replyQueue = properties.getReplyTo();
// 消息参数中取出关联id.
String correlationId = properties.getCorrelationId();
System.out.println("消息的回调队列为" + replyQueue + ", 关联id为" + correlationId);
int param = Integer.parseInt(new String(body));
String response = String.valueOf(fib(param));
AMQP.BasicProperties responseProperties = new AMQP.BasicProperties().builder().correlationId(correlationId).build();
// 将计算的结果发布给回调队列
channel.basicPublish("", replyQueue, responseProperties, response.getBytes());
// 确认消息已
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
}
private static int fib(int n) {
if (n == 0) {
return 0;
}
if (n == 1) {
return 1;
}
return fib(n - 1) + fib(n - 2);
}
}
- 客户端:
package com.pctf.advance;
import com.pctf.basic.ConnectionCreator;
import com.pctf.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.UUID;
public class RPCClient {
private Connection connection;
private Channel channel;
private String prcQueueName = "rpc_queue";
private String replyQueueName;
private Consumer consumer;
public RPCClient() throws IOException {
connection = ConnectionCreator.getConnection();
channel = connection.createChannel();
replyQueueName = channel.queueDeclare().getQueue();
consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String correlationId = properties.getCorrelationId();
System.out.println("该消息的关联id为:" + correlationId);
String result = new String(body);
System.out.println("远程调用结果为" + result);
// todo 对结果和关联的请求做处理逻辑
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(replyQueueName, false, consumer);
}
public void call(String message) throws Exception {
String uuid = UUID.randomUUID().toString();
// 设置关联id和回调队列
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().correlationId(uuid).replyTo(replyQueueName).build();
channel.basicPublish("rpc_exchange", "rpc_routing", properties, message.getBytes());
}
public void close() {
RabbitMQUtils.close(channel, connection);
}
}
先运行服务端,之后启动一个客户端并调用call()方法发送一个消息.
可以看到服务端处理消息并回调结果:
客户端收到回调,并处理:
对于没有处理过异步调用模型(不管是单进程还是集群环境)的同学来说,这个调用过程并不太好理解.其编程复杂性也比较高.
2. 持久化
持久化这个概念在之前多次提及，持久化可以提高RabbitMQ的可用性。本节将对持久化做一个总结。RabbitMQ的持久化分为三个部分：交换器持久化、队列的持久化和消息的持久化。
- 队列的持久化：声明队列时将durable参数设置为true，没有持久化的队列重启后将会丢失队列的元数据。
- 消息的持久化：队列的持久化并不能保证服务器重启后消息不丢失，需要将消息设置为持久化。通过将消息发布时的BasicProperties中的deliveryMode属性设置为2.
- 交换器的持久化： 声明交换器时将durable参数设置为true。
RabbitMQ的持久化机制并不是同步落到磁盘之中的，通过调用内核的fsync方法处理，可能消息仅仅保存到操作系统缓存之中而不是物理磁盘之中。如果服务器发生宕机或者进程崩溃，还可能发生消息丢失的问题。可以通过引入镜像队列机制，为关键业务设置镜像队列。
