RabbitMQ的RPC实战

心已赠人 2021-07-24 14:26 403阅读 0赞

**一 RabbitMQ官网实例**

[https://www.rabbitmq.com/getstarted.html][https_www.rabbitmq.com_getstarted.html]

**二 RabbitMQ的RPC代码位置**

[https://www.rabbitmq.com/tutorials/tutorial-six-java.html][https_www.rabbitmq.com_tutorials_tutorial-six-java.html]

[https://github.com/rabbitmq/rabbitmq-tutorials/tree/master/java][https_github.com_rabbitmq_rabbitmq-tutorials_tree_master_java]

**三 图例**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70][]

**四 代码**

1 客户端

package com.rabbitmq.RPC;
    
    import com.rabbitmq.client.*;
    import java.io.IOException;
    import java.util.UUID;
    import java.util.concurrent.TimeoutException;
    public class RPCClient implements AutoCloseable {
        private Connection connection;
        private Channel channel;
        private String requestQueueName = "rpc_queue";
        private static final String IP_ADDRESS = "127.0.0.1";
        private QueueingConsumer consumer;
        private String replyQueueName;
    
        public static void main( String[] argv ) throws IOException, TimeoutException, InterruptedException {
            RPCClient fibonacciRpc = new RPCClient();           // 初始化客户端RPC
            System.out.println(" [x] Requesting fib(30)");
            String response = fibonacciRpc.call("30");          // 远程调用30的斐波那契值，并得到最终结果
            System.out.println(" [.] Got '" + response + "'");
            fibonacciRpc.close();
        }
    
        public RPCClient() throws IOException, TimeoutException {       // 初始化客户端RPC的相关变量
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setUsername("root");
            factory.setPassword("123456");
            connection = factory.newConnection();
            channel = connection.createChannel();
            replyQueueName = channel.queueDeclare().getQueue();         // 回调队列，RPC服务端会将结果发送到replyQueueName队列
            consumer = new QueueingConsumer(channel);
            channel.basicConsume(replyQueueName,true,consumer);     // 客户端从replyQueueName队列取数据
        }
    
        public String call( String message ) throws IOException, InterruptedException {
            String response=null;
            final String corrId = UUID.randomUUID().toString();
    
            AMQP.BasicProperties props = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(corrId)
                    .replyTo(replyQueueName)
                    .build();
    
            // 发送消息到rpc_queue队列
            channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
    
            while(true){
                QueueingConsumer.Delivery delivery=consumer.nextDelivery();
                // 如果满足corrId相等，从replyQueueName队列取出计算结果
                if(delivery.getProperties().getCorrelationId().equals(corrId)){
                    response = new String(delivery.getBody());
                    break;
                }
            }
            return response;
        }
    
        public void close() throws IOException {
            connection.close();
        }
    }

2 服务端

package com.rabbitmq.RPC;
    
    import com.rabbitmq.client.*;
    import java.io.IOException;
    public class RPCServer {
        private static final String IP_ADDRESS = "127.0.0.1";
        private static final String RPC_QUEUE_NAME = "rpc_queue";
        private static int fib( int n ) {
            if (n == 0) return 0;
            if (n == 1) return 1;
            return fib(n - 1) + fib(n - 2);
        }
    
        public static void main( String[] argv ) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setUsername("root");
            factory.setPassword("123456");
    
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
            channel.queuePurge(RPC_QUEUE_NAME);
            channel.basicQos(1);
            System.out.println(" [x] Awaiting RPC requests");
            Consumer consumer  = new DefaultConsumer(channel){
                @Override
                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();
                    String response = "";
                    try {
                        // 接受从RPC_QUEUE_NAME队列的消息
                        String message = new String(body,"UTF-8");
                        int n = Integer.parseInt(message);
                        System.out.println("[.] fib("+message+")");
                        // 进行计算
                        response+=fib(n);
                        System.out.println(response);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }finally {
                        // 将计算结果发给回调队列，客户端会从回调队列取结果
                        channel.basicPublish("",properties.getReplyTo(),replyProps,response.getBytes("UTF-8"));
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            };
            channel.basicConsume(RPC_QUEUE_NAME,false,consumer);
        }
    }

**五 测试结果**

1 客户端

[x] Requesting fib(30)
    [.] Got '832040'

2 服务端

[x] Awaiting RPC requests
    [.] fib(30)
    832040

[https_www.rabbitmq.com_getstarted.html]: https://www.rabbitmq.com/getstarted.html
[https_www.rabbitmq.com_tutorials_tutorial-six-java.html]: https://www.rabbitmq.com/tutorials/tutorial-six-java.html
[https_github.com_rabbitmq_rabbitmq-tutorials_tree_master_java]: https://github.com/rabbitmq/rabbitmq-tutorials/tree/master/java
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5ncWl1bWluZw_size_16_color_FFFFFF_t_70]: /images/20210724/9f66feb6b1424e36bca1bd565274b646.png