RabbitMQ官网中文版教程：
http://rabbitmq.mr-ping.com/tutorials_with_python/[6]RPC.html
上述教程示例为pathon版，Java版及相应解释如下：
RPC客户端
package com.xc.rabbitmq.rpc;
import com.rabbitmq.client.*;
import java.util.UUID;
/**
* Created by xc.
*/
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");
factory.setPort(5672);
factory.setUsername("rabbit");
factory.setPassword("carrot");
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();
AMQP.BasicProperties props = new AMQP.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[] args) {
RpcClient client = null;
String response = null;
try {
client = new RpcClient();
System.out.println("RPCClient [x] Requesting fib(30)");
response = client.call("30");
System.out.println("RPCClient [.] Got '" + response + "'");
} catch (Exception e) {
e.printStackTrace();
} finally {
if (client != null) {
try {
client.close();
} catch (Exception e) {
;
}
}
}
}
}
RPC服务端
package com.xc.rabbitmq.rpc;
import com.rabbitmq.client.*;
/**
* Created by xc.
*/
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[] args) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("rabbit");
factory.setPassword("carrot");
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("RPCServer [x] Awaiting RPC requests");
while (true) {
String response = null;
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
BasicProperties props = delivery.getProperties();
AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
try {
String message = new String(delivery.getBody(), "UTF-8");
int n = Integer.parseInt(message);
System.out.println("RPCServer [.] fib(" + message + ")");
response = "" + fib(n);
} catch (Exception e) {
System.out.println(" [.] " + e.toString());
response = null;
} 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 e) {
;
}
}
}
}
}
先运行服务端程序，再运行客户端程序，结果如下：
注意：
1）消息属性（AMQP.BasicProperties）
AMQP协议给消息预定义了一系列的14个属性。大多数属性很少会用到，除了以下几个：
- delivery_mode（投递模式）：将消息标记为持久的（值为2）或暂存的（除了2之外的其他任何值）。第二篇教程里接触过这个属性，记得吧？
- content_type（内容类型）:用来描述编码的mime-type。例如在实际使用中常常使用application/json来描述JOSN编码类型。
- reply_to（回复目标）：通常用来命名回调队列。
- correlation_id（关联标识）：用来将RPC的响应和请求关联起来。
2） 如果我们接手到的消息的correlation_id是未知的，那就直接销毁掉它，因为它不属于我们的任何一条请求。
为什么我们接收到未知消息的时候不抛出一个错误，而是要将它忽略掉？这是为了解决服务器端有可能发生的竞争情况。尽管可能性不大，但RPC服务器还是有可能在已将应答发送给我们但还未将确认消息发送给请求的情况下死掉。如果这种情况发生，RPC在重启后会重新处理请求。这就是为什么我们必须在客户端优雅的处理重复响应，同时RPC也需要尽可能保持幂等性。
3）此处呈现的设计并不是实现RPC服务的唯一方式，但是他有一些重要的优势：
如果RPC服务器运行的过慢的时候，你可以通过运行另外一个服务器端轻松扩展它。试试在控制台中运行第二个 rpc_server.py 。
在客户端，RPC请求只发送或接收一条消息。不需要像 queue_declare 这样的异步调用。所以RPC客户端的单个请求只需要一个网络往返。
4）我们的代码依旧非常简单，而且没有试图去解决一些复杂（但是重要）的问题，如：
当没有服务器运行时，客户端如何作出反映。
客户端是否需要实现类似RPC超时的东西。
如果服务器发生故障，并且抛出异常，应该被转发到客户端吗？
在处理前，防止混入无效的信息（例如检查边界）
网友评论