基于消息中间件RabbitMQ实现简单的RPC服务

转载自基于消息中间件RabbitMQ实现简单的RPC服务

RPC(Remote Procedure Call,远程过程调用)，是一种计算机通信协议。对于两台机器而言，就是A服务器上的应用程序调用B服务器上的函数或者方法，由于不在同一个内存空间或机器上运行，因此需要借助于网络通信。

1. RPC框架

我们首先通过一张图理解RPC的工作流程：

因此，实现一个最简单的RPC服务，只需要Client、Server和Network，本文就是利用消息中间件RabbitMQ作为Network载体传输信息，实现简单的RPC服务。简单原理可如下图所示：

即：当Client发送RPC请求时，Client端是消息生产者，Server端是消息消费者；当Server返回结果时，Server端是消息生产者，Client是消息消费者；发送和返回使用不同的队列。

接下来我们通过代码，详细展示一个计算斐波那契数列的RPC服务。

2. RPCServer实现

2.1 Server初始化

/*** 队列名、交换机名、路由键*/
private static final String EXCHANGE_NAME = "rpc_exchange";
private static final String QUEUE_NAME = "request_rpc_queue";
private static final String ROUTING_KEY = "rpc_routing_key";private Connection connection = null;
private Channel channel = null;
private QueueingConsumer consumer = null;/*** Server的构造函数*/
private RPCServer() {try {//创建链接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Config.HOST);factory.setPort(Config.PORT);factory.setUsername(Config.USER);factory.setPassword(Config.PASSWORD);connection = factory.newConnection();//创建信道channel = connection.createChannel();//设置AMQP的通信结构channel.exchangeDeclare(EXCHANGE_NAME, "direct");channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);//设置消费者consumer = new QueueingConsumer(channel);channel.basicConsume(QUEUE_NAME, false, QUEUE_NAME, consumer);} catch (Exception e) {LOG.error("build connection failed!", e);}
}

初始化就是声明RabbitMQ的链接工厂、链接、信道、队列、交换机等等，并做了绑定，由此构成了AMQP的通信结构。

2.2 监听队列并反馈

/*** 开启server*/
private void startServer() {try {LOG.info("Waiting for RPC calls.....");while (true) {//获得文本消息QueueingConsumer.Delivery delivery = consumer.nextDelivery();BasicProperties props = delivery.getProperties();//返回消息的属性BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();long receiveTime = System.currentTimeMillis();JSONObject json = new JSONObject();try {String message = new String(delivery.getBody(), "UTF-8");int n = Integer.parseInt(message);LOG.info("Got a request: fib(" + message + ")");json.put("status", "success");json.put("result", fib(n));} catch (Exception e) {json.put("status", "fail");json.put("reason", "Not a Number!");LOG.error("receive message failed!", e);} finally {long responseTime = System.currentTimeMillis();json.put("calculateTime", (responseTime - receiveTime));channel.basicPublish("", props.getReplyTo(), replyProps, json.toString().getBytes("UTF-8"));channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}} catch  (Exception e) {LOG.error("server failed!", e);} finally {if (connection != null) {try {connection.close();} catch (Exception e) {LOG.error("close failed!", e);}}}
}

在该方法中使用了一个无限循环，每次处理一条消息。通过调用消费者对象的nextDelivery方法来获得RabbitMQ队列的最新一条消息。同时通过getProperties获取到消息中的反馈信息属性，用于标记客户端Client的属性。然后计算斐波那契数列的结果。最后通过basicAck使用消息信封向RabbitMQ确认了该消息。

到这里就实现了计算斐波那契数列RPC服务的Server端。

3. RPCClient实现

3.1 初始化CLient

/*** 消息请求的队列名、交换机名、路由键*/
private static final String EXCHANGE_NAME = "rpc_exchange";
private static final String QUEUE_NAME = "request_rpc_queue";
private static final String ROUTING_KEY = "rpc_routing_key";/*** 消息返回的队列名、交换机名、路由键*/
private static final String RESPONSE_QUEUE = "response_rpc_queue";
private static final String RESPONSE_ROUTING_KEY = "response_rpc_routing_key";/*** RabbitMQ的实体*/
private Connection connection = null;
private Channel channel = null;
private QueueingConsumer consumer = null;/*** 构造客户端* @throws Exception*/
private RPCClient() throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(Config.HOST);factory.setPort(Config.PORT);factory.setUsername(Config.USER);factory.setPassword(Config.PASSWORD);connection = factory.newConnection();channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct");channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);channel.queueDeclare(RESPONSE_QUEUE, false, false, false, null);channel.queueBind(RESPONSE_QUEUE, EXCHANGE_NAME, RESPONSE_ROUTING_KEY);consumer = new QueueingConsumer(channel);channel.basicConsume(RESPONSE_QUEUE, true, consumer);
}

这里声明AMQP结构体的方式和Server端类似，只不过Client端需要多声明一个队列，用于RPC的response。

3.2 发送/接收消息

/*** 请求server* @param message* @return* @throws Exception*/
private String requestMessage(String message) throws Exception {String response = null;String corrId = UUID.randomUUID().toString();BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(RESPONSE_QUEUE).build();channel.basicPublish("", QUEUE_NAME, props, message.getBytes("UTF-8"));while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();if (delivery.getProperties().getCorrelationId().equals(corrId)) {response = new String(delivery.getBody(),"UTF-8");break;}}return response;
}

BasicProperties用于存储你请求消息的属性，这里我设置了correlationId和replyTo属性，用于Server端的返回识别。