RabbitMQ学习笔记5-简单的RPC调用

利用空的queue名字("")让rabbitMQ生成一个唯一的队列名称，同时指定队列是：临时的(auto-delete)、私有的(exclusive)。

在发送的RPC调用消息里设置消息的属性(com.rabbitmq.client.AMQP.BasicProperties)的reply_to字段来传递上面那个随机的队列名称。

基本流程如下：

代码如下：

服务端RpcServer.java

package com.yzl.test4;

import java.io.IOException;

import org.apache.log4j.Logger;

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.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 基于rabbitMQ的RPC服务的服务提供者
 * 消息消费者角色，被动接收消息，然后回复消息
 * @author: yzl
 * @date: 2016-10-23
 */
public class RpcServer {
    //交换器名称
    private static final String EXCHANGE_NAME = "rpcDirectExchange";
    //client调用的消息存储queue
    private static final String QUEUE_NAME = "rpcQueue";
    //服务提供方的路由key，实际使用场景会是应用的appid或者appName
    private static final String SERVER_ROUTING_KEY = "rpc_server1";
    private static final Logger logger = Logger.getLogger(RpcServer.class);
    
    /**
     * @param args
     * @throws  
     * @throws IOException 
     */
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        
        //定义RPC的direct的交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //声明接收client调动请求的队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, SERVER_ROUTING_KEY);
        
        logger.info("服务启动完成，等待接收请求.....");
        
        //接收client端的请求
        channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                    throws IOException {
                //回复ack响应
                channel.basicAck(envelope.getDeliveryTag(), false);
                
                logger.info("收到client调用请求-----------");
                String msg = new String(body);
                logger.info("收到的消息如下:" + msg);
                String resp = null;
                
                //模拟解析需要调用的方法
                if(msg.indexOf("add") != -1){
                    String parameter = msg.substring(msg.indexOf("(")+1, msg.indexOf(")"));
                    String[] args = parameter.split(",");
                    resp = String.valueOf(add(Integer.valueOf(args[0]), Integer.valueOf(args[1])));
                }else{
                    resp = "method is not found!!";
                }
                logger.info("需要回调的client的queueName:" + properties.getReplyTo());
                try {
                    Integer time = new java.util.Random().nextInt(5000);
                    logger.info("休眠" + time + "毫秒");
                    //随即休眠，模拟服务调用方法正常业务方法耗时
                    Thread.sleep(time);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                //发送给默认的direct交换器，路由键为发送方的queueName,默认交换机会把消息转发给此路由键名称的queue
                channel.basicPublish("", properties.getReplyTo(), properties, (msg + ", resp is：" + resp).getBytes());
                logger.info("回复clent完成");
            }
        });
    }

    //模拟的server中的方法
    private static int add(int num1, int num2){
        logger.info("call add method, para num1 is :" + num1 + ",num2 is :" + num2);
        return num1 + num2;
    }
}

客户端

package com.yzl.test4;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.Queue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

/**
 * 基于rabbitMQ的RPC服务的服务调用者
 * 消费生产者角色，发送调用请求，然后读取回复
 * 发送的交换器 和 接收回应的交换器是不同的
 * @author: yzl
 * @date: 2016-10-23
 */
public class RpcClient {
    //交换器名称
    private static final String EXCHANGE_NAME = "rpcDirectExchange";
    //服务提供方的路由key
    private static final String SERVER_ROUTING_KEY = "rpc_server1";
    
    private static final Logger logger = Logger.getLogger(RpcClient.class);
    /**
     * @param args
     */
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        
        int threadCount = 1;
        
        //使用CountDownLatch控制10个线程一起运行
        final CountDownLatch cdl = new CountDownLatch(threadCount);
        
        //生成10个线程同时访问服务
        ExecutorService pool = Executors.newFixedThreadPool(threadCount);
        
        for(int i=0; i<threadCount; i++){
            final int index = i;
            pool.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        cdl.await();
                        
                        //Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments)
                        //默认direct exchange                      匿名的、               私有、自动删除
                        Queue.DeclareOk queue = channel.queueDeclare("", false, true, true, null);
                        //获取rabbit帮我们生成的随即的队列名称，来接收服务端返回的数据
                        String queueName = queue.getQueue();
                        
                        final String messageId = UUID.randomUUID().toString();
                        //定义replyTo属性 和 消息ID
                        BasicProperties props = new BasicProperties.Builder().replyTo(queueName).correlationId(messageId).build();
                        
                        logger.info("发送rpc调用请求，消息index:" + index);
                        
                        //发送RPC方法调用
                        channel.basicPublish(EXCHANGE_NAME, SERVER_ROUTING_KEY, props, ("add(" + index + "," + (index+1) + ")").getBytes());
                        
                        logger.info("等待服务器响应");
                        
                        //定义队列式的消费者处理器，之前是用的DefaultConsumer
                        QueueingConsumer consumer = new QueueingConsumer(channel);
                        
                        //把消费者处理器和队列对照起来
                        channel.basicConsume(queueName, consumer);
                        
                        //这里会堵塞，直到取到值或者超时
                        logger.info("尝试从consumer里取返回的值");
                        Delivery delivery = consumer.nextDelivery();
                        //Delivery delivery = consumer.nextDelivery(5000);
                        logger.info("成功取到消息，开始处理");
                        if(delivery.getProperties().getCorrelationId().equals(messageId)){
                            logger.info("收到服务器回复-----------");
                            String msg = new String(delivery.getBody());
                            logger.info("回复消息id：" + delivery.getProperties().getCorrelationId() + "内容:" + msg);
                        }
                        
                        /*
                        //该取消息的方式是异步的，不会堵塞，会导致后面的logger.info先于handleDelivery执行
                        channel.basicConsume(queueName, new DefaultConsumer(channel){
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope,
                                    BasicProperties properties, byte[] body) throws IOException {
                                if(properties.getCorrelationId().equals(messageId)){
                                    logger.info("收到服务器回复-----------");
                                    String msg = new String(body);
                                    logger.info("回复消息id：" + properties.getCorrelationId() + "内容:" + msg);
                                }
                            }
                        });
                        */
                        
                        logger.info("消息处理完成");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        for(int i=0; i<threadCount; i++){
            cdl.countDown();
        }
    }
}

先运行server后运行client，输入如下：

2016-10-23 21:22:38,961 [com.yzl.test4.RpcServer]-[INFO] 服务启动完成，等待接收请求.....
2016-10-23 21:22:42,414 [com.yzl.test4.RpcServer]-[INFO] 收到client调用请求-----------
2016-10-23 21:22:42,414 [com.yzl.test4.RpcServer]-[INFO] 收到的消息如下:add(0,1)
2016-10-23 21:22:42,414 [com.yzl.test4.RpcServer]-[INFO] call add method, para num1 is :0,num2 is :1
2016-10-23 21:22:42,414 [com.yzl.test4.RpcServer]-[INFO] 需要回调的client的queueName:amq.gen-iek4oYqWiqQ-HU7-i2g6mA
2016-10-23 21:22:42,414 [com.yzl.test4.RpcServer]-[INFO] 休眠3999毫秒
2016-10-23 21:22:46,415 [com.yzl.test4.RpcServer]-[INFO] 回复clent完成

2016-10-23 21:22:42,398 [com.yzl.test4.RpcClient]-[INFO] 发送rpc调用请求，消息index:0
2016-10-23 21:22:42,414 [com.yzl.test4.RpcClient]-[INFO] 等待服务器响应
2016-10-23 21:22:42,414 [com.yzl.test4.RpcClient]-[INFO] 尝试从consumer里取返回的值
2016-10-23 21:22:46,415 [com.yzl.test4.RpcClient]-[INFO] 成功取到消息，开始处理
2016-10-23 21:22:46,415 [com.yzl.test4.RpcClient]-[INFO] 收到服务器回复-----------
2016-10-23 21:22:46,415 [com.yzl.test4.RpcClient]-[INFO] 回复消息id：1e4b7fb3-1728-41e2-8cac-b13fe88d5a20内容:add(0,1), resp is：1
2016-10-23 21:22:46,415 [com.yzl.test4.RpcClient]-[INFO] 消息处理完成