博客

最近在项目中需要用到RabbitMQ的消息分发机制，Client端在发送消息给Server端处理之后还需要等待Server端的处理结果，开始很是困惑Server端如何将处理完成之后的结果再返回给相应发送这个消息的Client端，直到翻阅官方的资料才发现，Client端在发送消息的同时可以一并在 BasicProperties中将回调地址与collectionId一起发送到Server端，Server端在处理完成之后就可以通过这两个参数来将处理结果相应的再返回给发送此消息的Client端。

Client端与Server的端具体实现如下。

Client端：

package cn.muyi.RabbitMQ;

import java.io.IOException;
import java.util.UUID;

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.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

public class RPCClient {
    private static String requestQueueName = "rpc_queue";
    private static String replyQueueName;
    public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
        // 创建连接和频道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("你的服务端Host");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //若此处不声明队列，则消息发送者先运行而队列未声明时消息会丢失。
        //channel.queueDeclare(requestQueueName, false, false, false, null);
        replyQueueName = channel.queueDeclare().getQueue();
        String message = "Welcome MuYi";
        String collectionId = UUID.randomUUID().toString();
        //存入回调队列名与collectionId
        BasicProperties bpro = new BasicProperties().builder().correlationId(collectionId).replyTo(replyQueueName).build();
        channel.basicPublish("", requestQueueName, bpro, message.getBytes());
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //指定消费队列 关闭ack机制
        channel.basicConsume(replyQueueName, true, consumer);
        String responseMsg = null;
        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            if(collectionId.equals(delivery.getProperties().getCorrelationId())){
                responseMsg = new String(delivery.getBody());
                break;
            }
        }
        System.out.println("Received "+responseMsg);
    }
}

Server端：

package cn.muyi.RabbitMQ;

import java.io.IOException;

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.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

public class RpcReceiver {
    private static String requestQueueName = "rpc_queue";

    public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
        // 创建连接和频道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("你的服务端Host");
        Connection connection = factory.newConnection();
        //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
        Channel channel = connection.createChannel();
        channel.queueDeclare(requestQueueName, false, false, false, null);
        //创建队列消费者  
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //指定消费队列 打开ack机制
        channel.basicConsume(requestQueueName, false,consumer);
        while(true){
            //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("Received Client Message:"+message);
            //获取回调队列名与Correlation Id
            BasicProperties bpro = delivery.getProperties();
            String replName = bpro.getReplyTo();
            BasicProperties replBP = new BasicProperties().builder().correlationId(bpro.getCorrelationId()).build();
            String responseMsg = "Just Do It";
            channel.basicPublish("", replName, replBP, responseMsg.getBytes());
            //发送应答，通过delivery.getEnvelope().getDeliveryTag()获取此次确认的消息的序列号
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }

}

注：BasicProperties 中常用参数说明

Message properties

The AMQP protocol predefines a set of 14 properties that go with a message. Most of the properties are rarely used, with the exception of the following:

deliveryMode: Marks a message as persistent (with a value of 2) or transient (any other value). You may remember this property from the second tutorial.
contentType: Used to describe the mime-type of the encoding. For example for the often used JSON encoding it is a good practice to set this property to: application/json.
replyTo: Commonly used to name a callback queue.
correlationId: Useful to correlate RPC responses with requests