RabbitMQ学习：远程结果调用

Posted 2020-09-28 Stark_Tan

 

场景：我们需要在传输消息时得到结果

客服端在发送请求时会发送回调队列，服务端处理事情完成后会将结果返回到回调队列中，在增加关联标志关联每个请求和服务返回

客户端代码：

public class RPCClient {     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);         //为每一个客户端获取一个随机的回调队列         String replyQueueName = channel.queueDeclare().getQueue();         //为每一个客户端创建一个消费者（用于监听回调队列，获取结果）         QueueingConsumer consumer = new QueueingConsumer(channel);         //消费者与队列关联         channel.basicConsume(replyQueueName, true, consumer);         String response = null;         String corrId = java.util.UUID.randomUUID().toString();         //设置replyTo和correlationId属性值         AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();         //发送消息到rpc_queue队列         channel.basicPublish("", RPC_Queue_Name, props, "8".getBytes());         while (true) {             QueueingConsumer.Delivery delivery = consumer.nextDelivery();             if (delivery.getProperties().getCorrelationId().equals(corrId)) {                 response = new String(delivery.getBody(),"UTF-8");                 break;             }         }         System.out.println( "fib(8) is " + response);     } }

服务端代码：

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);     } }