精通RabbitMQ之RPC同步调用

本是古典何须时尚 2022-04-03 15:11 309阅读 0赞

### 精通RabbitMQ之RPC同步调用 ###

前面我们对应用解耦做过分析，我们能够使用消息中间件来完成应用解耦，很大一部分原因是因为我们的系统之间可以异步处理并且不关心结果回执。假如我们现在需要异步处理的结果，我们还可以使用消息队列吗？

实际上Rabbitmq是支持`Remote procedure call (RPC)`同步调用的。RabbitMQ RPC同步调用实际上是使用了两个异步调用完成的，生产者投递消息到消息队列的同时，自己也作为消费者等待reply队列的返回消息，消费者接受消息队列消息同时，也作为消息发送者发送返回消息到消息队列。

参考下图：

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

大家有兴趣的话可以，可以参考一下[官方文档][Link 1] ,这里不是我们的重点，如果没兴趣也可以完全不了解，毕竟我们现在有专门的微服务来做RPC，不需要`rabbitmq`，性能也不好。但是我们这里还是给出大家案例，大家结合代码比较好理解：

try { 
        //申明请求队列
        channel.queueDeclare("request", true, false, false, null);
        //申明返回队列
        channel.queueDeclare("response", true, false, false, null);
        //绑定
        channel.queueBind("test", "request", "request", null);
        channel.queueBind("test", "response", "response", null);
    
        //下面是生产者配置代码
        //correlationId - 每个请求独一无二的标识
        String correlationId = UUID.randomUUID().toString();
    
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                //指明请求Id
                .correlationId(correlationId)
                //指明返回消息队列
                .replyTo("response")
                .build();
    
        //发布request消息
        channel.basicPublish("test", "request", props, messageBodyBytes);
    
        //存储处理结果
        final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
    
        //监听返回队列拿结果
        channel.basicConsume("response", true, new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                if (properties.getCorrelationId().equals(correlationId)) { 
                    response.offer(new String(body, "UTF-8"));
                }
            }
        });
    
    
        //下面是消费者配置处理消息代码（放在一起比较好说明，大家测试的时候分开就好了）
        //设置一次处理一条消息
        channel.basicQos(0,1,true);
    
        System.out.println(" [x] Awaiting RPC requests");
    
        Channel finalChannel1 = channel;
        Consumer consumer = new DefaultConsumer(finalChannel1) { 
            @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 = "success";
                //处理结果发布到处理结果队列
                finalChannel1.basicPublish( "test", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
    
                //回执处理成功
                finalChannel1.basicAck(envelope.getDeliveryTag(), false);
    
                // RabbitMq consumer worker thread notifies the RPC server owner thread
                synchronized(this) { 
                    this.notify();
                }
            }
        };
    
        channel.basicConsume("response", false, consumer);
    
        // Wait and be prepared to consume the message from RPC client.
        while (true) { 
            synchronized(consumer) { 
                try { 
                    consumer.wait();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    } catch (IOException e) { 
        e.printStackTrace();
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQzMDAzNg_size_16_color_FFFFFF_t_70]: /images/20220403/ae80dbac699d410fa356742f5e6cbae3.png
[Link 1]: http://www.rabbitmq.com/tutorials/tutorial-six-java.html