建站
咨询
售后
建站咨询
这篇文章主要介绍了RabbitMQ如何实现RPC远程调用消息队列，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。
客户端接口
我们创建一个客户端类来说明如何使用RPC服务，暴露一个call方法来发送RPC请求和数据获取结果。
FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();
String result = fibonacciRpc.call("4");
System.out.println( "fib(4) is " + result);尽管RPC是编程中一种常见的模式，但其也常常饱受批评。因为程序员常常不知道调用的方法是本地方法还是一个RPC方法，这在调试中常常增加一些不必要的复杂性。我们应该简化代码，而不是滥用RPC导致代码变的臃肿。
回调队列
一般来说，通过RabbitMQ实现RPC非常简单，客户端发送一个请求消息，服务端响应消息就完成了。为了接收到响应内容，我们在请求中发送”callback“队列地址，也可以使用默认的队列。
callbackQueueName = channel.queueDeclare().getQueue();
BasicProperties props = new BasicProperties .Builder().replyTo(callbackQueueName) .build();
channel.basicPublish("", "rpc_queue", props, message.getBytes());AMQP协议中预定了14个消息属性，除了下面几个，其它的都很少使用：
deliveryMode ： 标识消息是持久化还是瞬态的。
contentType ： 描述 mime-type的编码类型，如JSON编码为”application/json“。
replyTo ： 通常在回调队列中使用。
correlationId ： 在请求中关联RPC响应时使用。
关联Id（Correlation Id）
在前面的方法中，要求在每个RPC请求创建回调队列，这可真是一件繁琐的事情，但幸运的是我们有个好方法-在每个客户端创建一个简单的回调队列。
这样问题又来了，队列如何知道这些响应来自哪个请求呢？这时候correlationId就出场了。我们在每个请求中都设置一个唯一的值，这样我们在回调队列中接收消息的时候就能知道是哪个请求发送的。如果收到未知的correlationId，就废弃该消息，因为它不是我们发出的请求。
你可能会问，为什么抛弃未知消息而不是抛出错误呢？这是由服务器竞争资源所导致的。尽管这不太可能，试想一下，如果RPC服务器在发送完响应后而在发送应答消息前死掉了，重启RPC服务器会重新发送请求。这就是我们在客户机上优雅地处理重复的反应，RPC应该是等同的。
（1）客户端启动，创建一个匿名且唯一的回调队列。
（2）对每个RPC请求，客户端发送一个包含replyTo和correlationId两个属性的消息。
（3）请求发送到rpc_queue队列。
（4）RPC服务在队列中等待请求，当请求出现时，根据replyTo字段使用队列将结果发送到客户端。
（5）客户端在回调队列中等待数据。当消息出现时，它会检查correlationId属性，如果该值匹配的话，就会返回响应结果给应用。
示例代码
RPCServer.java
package com.favccxx.favrabbit;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;
public class RPCServer {
private static final String RPC_QUEUE_NAME = "rpc_queue";
private static int fib(int n) {
if (n == 0)
return 0;
if (n == 1)
return 1;
return fib(n - 1) + fib(n - 2);
}
public static void main(String[] argv) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
channel.basicQos(1);
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
System.out.println(" [x] Awaiting RPC requests");
while (true) {
String response = null;
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
BasicProperties props = delivery.getProperties();
BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId())
.build();
try {
String message = new String(delivery.getBody(), "UTF-8");
int n = Integer.parseInt(message);
System.out.println(" [.] fib(" + message + ")");
response = "" + fib(n);
} catch (Exception e) {
System.out.println(" [.] " + e.toString());
response = "";
} finally {
channel.basicPublish("", props.getReplyTo(), replyProps, response.getBytes("UTF-8"));
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
if (connection != null) {
try {
connection.close();
} catch (Exception ignore) {
}
}
}
}
}RPCClient.java
package com.favccxx.favrabbit;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;
import java.util.UUID;
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("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
replyQueueName = channel.queueDeclare().getQueue();
consumer = new QueueingConsumer(channel);
channel.basicConsume(replyQueueName, true, consumer);
}
public String call(String message) throws Exception {
String response = null;
String corrId = UUID.randomUUID().toString();
BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();
channel.basicPublish("", requestQueueName, 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;
}
public void close() throws Exception {
connection.close();
}
public static void main(String[] argv) {
RPCClient fibonacciRpc = null;
String response = null;
try {
fibonacciRpc = new RPCClient();
System.out.println(" [x] Requesting fib(30)");
response = fibonacciRpc.call("30");
System.out.println(" [.] Got '" + response + "'");
} catch (Exception e) {
e.printStackTrace();
} finally {
if (fibonacciRpc != null) {
try {
fibonacciRpc.close();
} catch (Exception ignore) {
}
}
}
}
}先启动RPCServer，然后运行RPCClient,控制台输出如下内容
RPCClient[x] Requesting fib(30) RPCClient[.] Got '832040' RPCServer[x] Awaiting RPC requests RPCServer[.] fib(30) |
感谢你能够认真阅读完这篇文章，希望小编分享的“RabbitMQ如何实现RPC远程调用消息队列”这篇文章对大家有帮助，同时也希望大家多多支持创新互联，关注创新互联行业资讯频道，更多相关知识等着你来学习!
另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。
网站名称：RabbitMQ如何实现RPC远程调用消息队列-创新互联
网页路径：http://lswzjz.com/article/gipds.html
