（转） RabbitMQ学习之远程过程调用(RPC)（java）

http://blog.csdn.net/zhu_tianwei/article/details/40887885

在一般使用RabbitMQ做RPC很容易。客户端发送一个请求消息然后服务器回复一个响应消息。为了收到一个响应，我们需要发送一个'回调'的请求的队列地址。我们可以使用默认队列（在Java客户端除外）。

AMQP协议给消息定义了14个属性。大部分的属性很少使用，除了下面几个：
deliveryMode: 将消息标记为持久（值为2）或瞬态（任何其他值）。你可能记得在第二个教程中使用了这个属性。
contentType:用来设置mime类型。例如经常使用的JSON格式数据，就需要将此属性设置为：application/json。
replyTo: 通常用来命名一个回调队列.
correlationId: 用来关联RPC请求的响应.

RPC工作流程:

1)、客户端启动时，创建了一个匿名的回调队列。
2)、在一个RPC请求中，客户端发送一个消息，它有两个属性：1.REPLYTO，用来设置回调队列名；2.correlationId，对于每个请求都被设置成唯一的值。
3)、请求被发送到rpc_queue队列.
4)、RPC工作者（又名：服务器）等待接收该队列的请求。当收到一个请求，它就会处理并把结果发送给客户端，使用的队列是replyTo字段指定的。
5)、客户端等待接收回调队列中的数据。当接到一个消息，它会检查它的correlationId属性。如果它和设置的相匹配，就会把响应返回给应用程序。

1、RPC服务器的RPCServer.java，接收消息调用rpc并返回结果

[java] view plaincopy print?

package cn.slimsmart.rabbitmq.demo.rpc;
import java.security.MessageDigest;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
//RPC调用服务端
public class RPCServer {
private static final String RPC_QUEUE_NAME = "rpc_queue";
public static void main(String[] args) throws Exception {
//• 先建立连接、通道，并声明队列
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.36.217");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(AMQP.PROTOCOL.PORT);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
//•可以运行多个服务器进程。通过channel.basicQos设置prefetchCount属性可将负载平均分配到多台服务器上。
channel.basicQos(1);
QueueingConsumer consumer = new QueueingConsumer(channel);
//打开应答机制autoAck=false
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
System.out.println(" [x] Awaiting RPC requests");
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
BasicProperties props = delivery.getProperties();
BasicProperties replyProps = new BasicProperties.Builder()
.correlationId(props.getCorrelationId()).build();
String message = new String(delivery.getBody());
System.out.println(" [.] getMd5String(" + message + ")");
String response = getMd5String(message);
//返回处理结果队列
channel.basicPublish("", props.getReplyTo(), replyProps,
response.getBytes());
//发送应答
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
// 模拟RPC方法获取MD5字符串
public static String getMd5String(String str) {
MessageDigest md5 = null;
try {
md5 = MessageDigest.getInstance("MD5");
} catch (Exception e) {
System.out.println(e.toString());
e.printStackTrace();
return "";
}
char[] charArray = str.toCharArray();
byte[] byteArray = new byte[charArray.length];
for (int i = 0; i < charArray.length; i++)
byteArray[i] = (byte) charArray[i];
byte[] md5Bytes = md5.digest(byteArray);
StringBuffer hexValue = new StringBuffer();
for (int i = 0; i < md5Bytes.length; i++) {
int val = ((int) md5Bytes[i]) & 0xff;
if (val < 16)
hexValue.append("0");
hexValue.append(Integer.toHexString(val));
}
return hexValue.toString();
}
}

2.客户端RPCClient.java，发送rpc调用消息，接收结果

[java] view plaincopy print?

package cn.slimsmart.rabbitmq.demo.rpc;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;
//RPC调用客户端
public class RPCClient {
private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
private String replyQueueName;
private QueueingConsumer consumer;
public RPCClient() throws Exception {
//• 先建立一个连接和一个通道，并为回调声明一个唯一的'回调'队列
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.36.217");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(AMQP.PROTOCOL.PORT);
connection = factory.newConnection();
channel = connection.createChannel();
//• 注册'回调'队列，这样就可以收到RPC响应
replyQueueName = channel.queueDeclare().getQueue();
consumer = new QueueingConsumer(channel);
channel.basicConsume(replyQueueName, true, consumer);
}
//发送RPC请求
public String call(String message) throws Exception {
String response = null;
String corrId = java.util.UUID.randomUUID().toString();
//发送请求消息，消息使用了两个属性：replyto和correlationId
BasicProperties props = new BasicProperties.Builder()
.correlationId(corrId).replyTo(replyQueueName).build();
channel.basicPublish("", requestQueueName, props, message.getBytes());
//等待接收结果
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
//检查它的correlationId是否是我们所要找的那个
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
response = new String(delivery.getBody());
break;
}
}
return response;
}
public void close() throws Exception {
connection.close();
}
}

3、运行client主函数RPCMain.java

[java] view plaincopy print?

package cn.slimsmart.rabbitmq.demo.rpc;
public class RPCMain {
public static void main(String[] args) throws Exception {
RPCClient rpcClient = new RPCClient();
System.out.println(" [x] Requesting getMd5String(abc)");
String response = rpcClient.call("abc");
System.out.println(" [.] Got '" + response + "'");
rpcClient.close();
}
}

先运行服务端，再运行RPCMain，发送消息调用RPC。

这里介绍的是该设计不是实现RPC服务的唯一可能，但它有一些重要的优点:
1）如果RPC服务器速度太慢，你可以通过运行多个RPC服务器。尝试在一个新的控制台上运行第二RPCServer。
2）RPC客户端只发送和接收一个消息。不需要queueDeclare那样要求同步调用。因此，RPC客户端只需要在一个网络上发送和接收为一个单一的RPC请求。

转载于:https://www.cnblogs.com/telwanggs/p/7124709.html

（转） RabbitMQ学习之远程过程调用(RPC)（java）相关推荐

（转）RabbitMQ学习之主题topic（java）
http://blog.csdn.net/zhu_tianwei/article/details/40887775 参考:http://blog.csdn.NET/lmj623565791/artic ...
（转） RabbitMQ学习之发布/订阅（java）
http://blog.csdn.net/zhu_tianwei/article/details/40887733 参考:http://blog.csdn.NET/lmj623565791/artic ...
[译]RabbitMQ教程C#版 - 远程过程调用(RPC)
先决条件本教程假定 RabbitMQ 已经安装,并运行在localhost标准端口(5672).如果你使用不同的主机.端口或证书,则需要调整连接设置. 从哪里获得帮助如果您在阅读本教程时遇到困难, ...
RabbitMQ远程过程调用(RPC)
RabbitMQ远程过程调用(RPC) 准备环境使用Pika RabbitMQ客户端,客户端版本(1.0.0以上) 客户端接口示例 FibonacciRpcClient是一个简单的客户端类,它暴露了 ...
（转） RabbitMQ学习之helloword（java）
http://blog.csdn.net/zhu_tianwei/article/details/40835555 amqp-client:http://www.rabbitmq.com/java-c ...
RabbitMQ学习总结第一篇：理论篇
目录 RabbitMQ学习总结第一篇:理论篇 RabbitMQ学习总结第二篇:快速入门HelloWorld RabbitMQ学习总结第三篇:工作队列Work Queue RabbitMQ学习总结 ...
RabbitMQ学习笔记（3）----RabbitMQ Worker的使用
1. Woker队列结构图这里表示一个生产者生产了消息发送到队列中,但是确有两个消费者在消费同一个队列中的消息. 2. 创建一个生产者 Producer如下: package com.wangx.r ...
java项目_值得学习和练手的Java企业级开源项目，强烈推荐！
热门文章推荐:100本纸质书籍免费包邮!22款高颜值的终端工具,点燃了小姐姐对编程的热爱! 一条 SQL 引发的事故,同时小姐姐直接被开除!! 11月全国程序员平均工资新鲜出炉,内附封面图! 文末有彩 ...
RabbitMQ学习
RabbitMQ学习 1.概述用于进程通信的中间件. 优势: 劣势: 1.应用解耦:提高了系统容错性和可维护性 1.系统依赖越多不能保证MQ的高可用 2.异步提速:提升用户体验和系统吞吐量 2.复杂 ...

（转） RabbitMQ学习之远程过程调用(RPC)（java）

（转） RabbitMQ学习之远程过程调用(RPC)（java）相关推荐

最新文章

热门文章