RabbitMQ进阶(二)--RPC、持久化.md

本节继续探讨一些RabbitMQ的高阶用法以及应用场景。并结合之前的实例给出RabbitMQ的消息传输保障层级。

1. RabbitMQ 实现RPC调用

RPC即Remote Procedure Call，即远程过程调用。概念可参考 远程过程调用协议。
PRC协议有很多，比如CORBA、Java RMI、Thrift、Restful API等。
我试着总结了下RabbitMQ实现RPC的几个特点:

1. 实现异步的远程调用
2. 可以暂存请求,缓解服务端压力

3. 编程复杂,并且由于加入了新的节点,可能出现重复调用的问题.
   其实质可以看做是多生产者-多消费者-双队列的一个模型,在集群环境下有一定的应用场景.

   
   

根据上图所示,RPC的处理流程如下:

1. 客户端启动时,创建一个匿名的回调队列
2. 客户端为PRC请求设置两个属性:replyTo告之RPC服务端恢复请求时的回调队列, correlationId用来标记关联id
3. 请求发送到rpc_queue中
4. 服务端监听rpc_queue,处理请求,并将结果发送给reply_to设定的回调队列.
5. 客户端监听回调队列.检查关联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方法处理，可能消息仅仅保存到操作系统缓存之中而不是物理磁盘之中。如果服务器发生宕机或者进程崩溃，还可能发生消息丢失的问题。可以通过引入镜像队列机制，为关键业务设置镜像队列。