场景：我们需要在传输消息时得到结果
客服端在发送请求时会发送回调队列，服务端处理事情完成后会将结果返回到回调队列中，在增加关联标志关联每个请求和服务返回
客户端代码：
|
public class RPCClient { private final static String RPC_Queue_Name public static void main(String[] ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = Channel channel = //声明队列 channel.queueDeclare(RPC_Queue_Name, false, false, false, null); //为每一个客户端获取一个随机的回调队列 String //为每一个客户端创建一个消费者（用于监听回调队列，获取结果） QueueingConsumer //消费者与队列关联 channel.basicConsume(replyQueueName, String response = null; String corrId = java.util.UUID.randomUUID().toString(); //设置replyTo和correlationId属性值 AMQP.BasicProperties //发送消息到rpc_queue队列 channel.basicPublish("", RPC_Queue_Name, props, "8".getBytes()); while (true) { QueueingConsumer.Delivery if (delivery.getProperties().getCorrelationId().equals(corrId)) response = new String(delivery.getBody(),"UTF-8"); break; } } System.out.println( "fib(8) is } } |
服务端代码：
public class RPCServer {
private final static String RPC_Queue_Name = "rpc_queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel 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) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
//获取请求中的correlationId属性值，并将其设置到结果消息的correlationId属性中
BasicProperties props = delivery.getProperties();
AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
//获取回调队列名字
String callQueueName = props.getReplyTo();
String message = new String(delivery.getBody(),"UTF-8");
System.out.println(" [.] fib(" + message + ")");
//获取结果
String response = "" + fib(Integer.parseInt(message));
//先发送回调结果
channel.basicPublish("", callQueueName, replyProps,response.getBytes());
//后手动发送消息反馈
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
private static int fib(int i)
{
if(i==0) return 0;
if (i==1) return 1;
return fib(i-1) +fib(i-2);
}
}
|