RabbitMQ学习笔记（四）RPC

朱雀 2021-01-24 19:02 659阅读 0赞

项目地址：[https://github.com/gongxianshengjiadexiaohuihui/RabbitMQ/tree/master/RPC\_RabbitMQ][https_github.com_gongxianshengjiadexiaohuihui_RabbitMQ_tree_master_RPC_RabbitMQ]

RPC远程服务调用，举个例子就是客户端远程调用服务端的方法帮自己运算，并把结果返回

流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70][]

项目结构

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 1][]

Client

*  创建一个反馈队列，这个队列的作用是等待服务器返回处理结果
 *  发送请求，请求的内容包含，待处理的信息，correlaitonId(用来检查反馈内容是否是自己需要 的)，反馈队列的名字
 *  从反馈队列中取出信息，对信息进行验证（correlationId）,验证通过，返回结果

package com.ggp;
    
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.UUID;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @ClassName Client
     * @Description TODO
     * @Author Mr.G
     * @Date 2018/11/21 16:39
     * @Version 1.0
     */
    public class Client {
        private Connection connection;
        private Channel channel;
        /**
         * 请求队列名
         */
        private String requestQueueName = "my_queue";
        /**
         * 等待响应队列名
         */
        private String replyQueueName;
        private QueueingConsumer consumer;
    
        public Client() throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
    
            /**
             * 监听反馈队列的状态
             */
            replyQueueName = channel.queueDeclare().getQueue();
            consumer = new QueueingConsumer(channel);
            channel.basicConsume(replyQueueName,true,consumer);
    
        }
        public String call(String message) throws IOException, InterruptedException{
            String result;
            String corrID = UUID.randomUUID().toString();
            /**
             * 建造者模式，创建配置文件，把corrID和需要反馈信息的队列名放进去
             */
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().correlationId(corrID).replyTo(replyQueueName).build();
            channel.basicPublish("",requestQueueName,properties,message.getBytes());
            while (true){
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                if(delivery.getProperties().getCorrelationId().equals(corrID)){
                    result = new String(delivery.getBody(), "UTF-8");
                    break;
                }
            }
            return result;
        }
    
        public void close() throws  Exception{
            connection.close();
        }
        public static void main(String[] args)throws Exception{
            Client client = new Client();
            String result = client.call("20");
            System.out.println(result);
            client.close();
        }
    }

Server

*  接收客户端信息，处理内容，计算结果
 *  通过接受的信息，得到反馈队列的名字，给该队列发送信息，信息内容包括correlationId和处理结果
````
    package com.ggp;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.ConnectException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @ClassName Server
     * @Description TODO
     * @Author Mr.G
     * @Date 2018/11/21 17:10
     * @Version 1.0
     */
    public class Server {
        private static final String RPC_QUEUE_NAME = "my_queue";
        private static int fib(int n){
            if(n == 0){
                return 0;
            }
            if(n == 1){
                return 1;
            }
            return fib(n-1) + fib(n -1);
        }
        public Server() throws IOException, TimeoutException, InterruptedException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            Connection connection = connectionFactory.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("Server is waiting request");
            while (true){
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().correlationId(delivery.getProperties().getCorrelationId()).build();
                String message = new String(delivery.getBody(),"UTF-8");
                int n = Integer.parseInt(message);
                System.out.println("receive the message : "+n);
    
                String response = ""+fib(n);
                channel.basicPublish("",delivery.getProperties().getReplyTo(),properties,response.getBytes());
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }
        }
    }
````
测试方法
````
    package com.ggp.test;
    
    import com.ggp.Client;
    
    /**
     * @ClassName ClintTest
     * @Description TODO
     * @Author Mr.G
     * @Date 2018/11/22 8:41
     * @Version 1.0
     */
    public class ClintTest {
        public static void main(String[] args)throws Exception{
            Client client = new Client();
            String result = client.call("20");
            System.out.println(result);
    
        }
    }
````
````
    package com.ggp.test;
    
    import com.ggp.Client;
    import com.ggp.Server;
    
    /**
     * @ClassName Test
     * @Description TODO
     * @Author Mr.G
     * @Date 2018/11/22 8:39
     * @Version 1.0
     */
    public class ServerTest {
        public static void main(String[] args)throws Exception{
            new Server();
        }
    }
````
测试结果

![20181122104948892.png][]

![20181122104956988.png][]

[https_github.com_gongxianshengjiadexiaohuihui_RabbitMQ_tree_master_RPC_RabbitMQ]: https://github.com/gongxianshengjiadexiaohuihui/RabbitMQ/tree/master/RPC_RabbitMQ
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70]: /images/1611514961012.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 1]: /images/1611514941372.png
[20181122104948892.png]: /images/1611514920828.png
[20181122104956988.png]: /images/1611514909512.png