《RabbitMQ系列教程-第四章-06-RabbitMQ工作模式之RPC模式》

矫情吗；* 2023-01-02 03:25 203阅读 0赞

# 教程说明 #

*  本系列教程目录大纲：[《RabbitMQ系列教程-目录大纲》][RabbitMQ_-]

--------------------

*  本系列教程配套代码：[https://gitee.com/Horizon1024/rabbitmt.git][https_gitee.com_Horizon1024_rabbitmt.git]（码云地址）

--------------------

# RabbitMQ工作模式之RPC模式 #

### 4.6.1 简介 ###

前面几种模式的通信都是基于`Producer`发送消息到`Consumer`，然后`Consumer`进行消费，假设我们需要`Consumer`操作完毕之后返回给`Producer`一个回调呢？前面几种模式就行不通了；

例如我们要做一个远程调用加钱操作，客户端远程调用服务端进行加钱操作，操作完毕之后服务端将用户最新的余额返回给客户端；客户端进行后续操作，例如更新到数据库等；

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JiMTUwNzAwNDc3NDg_size_16_color_FFFFFF_t_70_pic_center]

在RPC模式中，客户端和服务器都是`Producer`也都是`Consumer`；

> RPC模式官网介绍：https://www.rabbitmq.com/tutorials/tutorial-five-java.html

### 4.6.2 客户端 ###

package com.lscl.rabbitmq;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.UUID;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeoutException;
    
    public class RPCClient implements AutoCloseable { 
    
        public Connection connection;
        public Channel channel;
        public String requestQueueName = "rpc_queue";
    
        public RPCClient() throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.40.132");
            factory.setPort(5672);
            factory.setUsername("lscl");
            factory.setPassword("admin");
            factory.setVirtualHost("/lscl");
    
            connection = factory.newConnection();
            channel = connection.createChannel();
        }
    
        public static void main(String[] argv) throws Exception { 
            // 初始化信息
            RPCClient rpcClient = new RPCClient();
            // 发起远程调用
            Integer response = rpcClient.call(20);
    
            System.out.println(response);
    
            rpcClient.channel.close();
            rpcClient.connection.close();
        }
    
        public Integer call(Integer money) throws IOException, InterruptedException { 
    
            // 随机生成一个correlationId
            final String corrId = UUID.randomUUID().toString();
    
            // 服务端回调的队列名(随机生成)
            String replyQueueName = channel.queueDeclare().getQueue();
            
            // 设置发送消息的一些参数
            AMQP.BasicProperties props = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(corrId)
                    .replyTo(replyQueueName)
                    .build();
    
    
            channel.basicPublish("", requestQueueName, props, (money + "").getBytes("UTF-8"));
    
            final BlockingQueue<Integer> response = new ArrayBlockingQueue<>(1);
    
            channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) { 
    
                // 回调方法,当收到消息之后,会自动执行该方法
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
    
                    if (properties.getCorrelationId().equals(corrId)) { 
                        System.out.println("响应的消息：" + new String(body));
    
                        response.offer(Integer.parseInt(new String(body, "UTF-8")));
                    }
                }
            });
    
            return response.take();
        }
    
        public void close() throws IOException { 
            connection.close();
        }
    }

### 4.6.2 服务端 ###

package com.lscl.rabbitmq;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class RPCServer { 
    
        private static final String RPC_QUEUE_NAME = "rpc_queue";
    
        // 总金额
        private static Integer money = 0;
    
        /** * 加钱方法 * @param n * @return */
        private static Integer addMoney(int n) { 
            money += n;
            return money;
        }
    
        public static void main(String[] argv) throws Exception { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.40.132");
            factory.setPort(5672);
            factory.setUsername("lscl");
            factory.setPassword("admin");
            factory.setVirtualHost("/lscl");
    
            try (
                    Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel()
            ) { 
    
                channel.queueDeclare(RPC_QUEUE_NAME, true, false, false, null);
    
                System.out.println("等待客户端请求.....");
    
                while (true) { 
    
                    // 接受到客户端的请求(消息)
                    channel.basicConsume(RPC_QUEUE_NAME, true, new DefaultConsumer(channel) { 
    
                        // 回调方法,当收到消息之后,会自动执行该方法
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                            AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                                    .Builder()
                                    .correlationId(properties.getCorrelationId())
                                    .build();
    
                            System.out.println("客户端的消息: "+new String(body,"UTF-8"));
                            String response = "";
    
                            try { 
                                String message = new String(body, "UTF-8");
    
                                // 调用加钱方法
                                response = addMoney(Integer.parseInt(message)) + "";
    
                            } finally { 
    
                                // 发送一个消息给客户端
                                /* properties.getReplyTo(): Client端设置的回调队列名 replyProps:封装的参数(主要是CorrelationId) */
                                channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
                            }
                        }
                    });
                }
    
            }
        }
    }

### 4.6.3 RPC模式小结 ###

严格意义上来说RPC并不是一种新的交换模式，他其实还是借助原有的模式（上述案例是采用`Work`模式）来达到一些不同的功能，在RPC模式中只有客户端和服务端，并且客户端和服务端都既是`Producer`也是`Consumer`；

在RPC模式中，使用到了`CorrelationId`来解决回调队列问题，如果没有`CorrelationId`则需要为每一个PRC请求创建一个单一的回调队列，效率较低，有了`CorrelationId`之后，一个客户端永远使用`CorrelationId`对应的队列；

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JiMTUwNzAwNDc3NDg_size_16_color_FFFFFF_t_70_pic_center 1]

[RabbitMQ_-]: https://blog.csdn.net/Bb15070047748/article/details/112194912
[https_gitee.com_Horizon1024_rabbitmt.git]: https://gitee.com/Horizon1024/rabbitmt.git
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JiMTUwNzAwNDc3NDg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/067aae8d07fc4e38a97f4fd5e5c07db6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JiMTUwNzAwNDc3NDg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/953b3d71b30a464d8d3239d1967b4d95.png