网站建设中哪些最重要性,国内专门做情侣的网站商城,金山软件有哪些产品,网站增加权重吗转载自 基于消息中间件RabbitMQ实现简单的RPC服务 RPC(Remote Procedure Call,远程过程调用)#xff0c;是一种计算机通信协议。对于两台机器而言#xff0c;就是A服务器上的应用程序调用B服务器上的函数或者方法#xff0c;由于不在同一个内存空间或机器上运行#xff0c…转载自 基于消息中间件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端的返回识别。
4. 运行测试
Client端发送 Server端接收并处理
Client收到计算结果
由于我运行RabbitMQ的服务器是租用的阿里云的差不多传输时延在60ms左右如果把RPC服务和消息中间件同机房部署的话延时基本上就在ms级别。
5. FAQ
5.1 说明
需要体验完整的过程你需要如下环境
JDK1.6以上 Maven RabbitMQ5.2 源代码
完整代码代码请戳github
其中Server的代码在
rpc.RPCServerClient端的代码位置
rpc.RPCClient
