RabbitMQ中的RPC实现

2022-09-27 18:41:43

1、RPC简述

RPC，Remote Procedure Call 远程过程调用。通俗讲，两段程序不在同一个内存空间，无法直接通过方法名调用，就需要通过网络通信方式调用。对于RabbitMQ，本身就是用于消息通信。简单的RabbitMQ是，生产端发送消息，经由交换器，到达队列。消费端不需要知道生产端，消费端订阅队列，消费队列中的消息。而对于RPC，希望消费端消费消息后，返回一个结果，结果经由网络，再返回给生产端。

不考虑RabbitMQ针对RPC的特有设计。最简单的设计是，生产端和消费端共同约定消费队列和回复队列，同时生产端每次发送消息时指定一个唯一ID。生产端将消息和唯一ID发送给消费队列，消费者从消费队列获取消息。处理后，将结果和生产端发送过来的唯一ID，发送给回复队列。生产端从回复队列获取消息和ID，判断ID是否匹配，匹配，则此消息为回复消息。

以上实现的RPC存在问题：生产端和消费端需要约定回复队列，这就要求生产端和消费端互相知道，这无法实现解耦。解决方案：生产端在发送消息时，也将回复队列名称随消息一起发送给队列。

2、举例说明RabbitMQ中的RPC实现

相关要点，见源码注释

pom依赖

   <dependencies>

        <dependency>

            <groupId>com.rabbitmq</groupId>

            <artifactId>amqp-client</artifactId>

            <version>4.6.0</version>

        </dependency>

    </dependencies>

生产端，也就是客户端代码

 package test;

 import java.io.IOException;

 import java.util.UUID;

 import com.rabbitmq.client.AMQP;

 import com.rabbitmq.client.AMQP.BasicProperties;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import utils.ChannelUtils;

 public class RPCClient {

     public static void main(String[] args) throws IOException {

         RPCClient rpcClient = new RPCClient();

         rpcClient.client();

     }

     public void client() throws IOException {

         //此方法封装了如何连接RabbitMQ和创建connection，channel.源码见附录

         Channel channel = ChannelUtils.getChannelInstance("client");

         channel.exchangeDelete("exchange_rpc");

         channel.exchangeDeclare("exchange_rpc", "direct", false, false, null);

         channel.queueDelete("queue_rpc");

         channel.queueDeclare("queue_rpc", false, false, false, null);

         channel.queueBind("queue_rpc", "exchange_rpc", "rpc");

         //此处注意：我们声明了要回复的队列。队列名称由RabbitMQ自动创建。

         //这样做的好处是：每个客户端有属于自己的唯一回复队列，生命周期同客户端

         String replyQueue = channel.queueDeclare().getQueue();

         final String corrID = UUID.randomUUID().toString();

         //这里我们设计三类消息。

         //消息1：指定回复队列和ID

         //消息2：仅指定回复队列

         //消息3：不指定回复队列和ID

         AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();

         // 指定回复队列和回复correlateId

         builder.replyTo(replyQueue).correlationId(corrID);

         AMQP.BasicProperties properties = builder.build();

         for (int i = 0; i < 2; i++) {

             channel.basicPublish("exchange_rpc", "rpc", properties,

                     (System.currentTimeMillis() + "-rpc发送消息1").getBytes());

         }

         AMQP.BasicProperties.Builder builder2 = new AMQP.BasicProperties.Builder();

         // 指定回复队列，未指定回复correlateId

         builder2.replyTo(replyQueue);

         AMQP.BasicProperties properties2 = builder2.build();

         for (int i = 0; i < 2; i++) {

             channel.basicPublish("exchange_rpc", "rpc", properties2,

                     (System.currentTimeMillis() + "-rpc发送消息2").getBytes());

         }

         for (int i = 0; i < 2; i++) {

             // 未指定回复队列和correlateId

             channel.basicPublish("exchange_rpc", "rpc", null, (System.currentTimeMillis() + "-rpc发送消息3").getBytes());

         }

         DefaultConsumer c = new DefaultConsumer(channel) {

             //这是一个回调函数，客户端获取消息，就调用此方法，处理消息

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)

                     throws IOException {

                 if (corrID.equals(properties.getCorrelationId())) {

                     System.out.println("correlationID对应上的消息：" + new String(body));

                 } else {

                     System.out.println("correlationID未对应上的消息：" + new String(body));

                 }

             }

         };

         channel.basicConsume(replyQueue, true, c);

     }

 }

消费端，也就是服务器端代码

 package test;

 import java.io.IOException;

 import com.rabbitmq.client.AMQP;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import com.rabbitmq.client.AMQP.BasicProperties;

 import utils.ChannelUtils;

 public class RPCServer {

     public static void main(String[] args) throws IOException {

         RPCServer rpcServer = new RPCServer();

         rpcServer.Server();

     }

     public void Server() throws IOException {

         final Channel channel = ChannelUtils.getChannelInstance("server");

         DefaultConsumer c = new DefaultConsumer(channel) {

             //这是一个回到函数，服务器端获取到消息，就会调用此方法处理消息

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)

                     throws IOException {

                 System.out.println(new String(body));

                 AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();

                 //我们在将要回复的消息属性中，放入从客户端传递过来的correlateId

                 builder.correlationId(properties.getCorrelationId());

                 AMQP.BasicProperties prop = builder.build();

                 //发送给回复队列的消息，exchange=""，routingKey=回复队列名称

                 //因为RabbitMQ对于队列，始终存在一个默认exchange=""，routingKey=队列名称的绑定关系

                 channel.basicPublish("", properties.getReplyTo(), prop, (new String(body) + "-回复").getBytes());

             }

         };

         channel.basicConsume("queue_rpc", true, c);

     }

 }

附录：ChannelUtils 的源码

package utils;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

public class ChannelUtils {

    // AMQP的连接其实是对Socket做的封装, 注意以下AMQP协议的版本号，不同版本的协议用法可能不同。

    public static Channel getChannelInstance(String ConnectionDescription) {

        try {

            ConnectionFactory connectionFactory = getConnectionFactory();

            Connection connection = connectionFactory.newConnection(ConnectionDescription);

            return connection.createChannel();

        } catch (Exception e) {

            throw new RuntimeException("获取Channel连接失败");

        }

    }

    public static ConnectionFactory getConnectionFactory() {

        ConnectionFactory connectionFactory = new ConnectionFactory();

        connectionFactory.setHost("192.168.1.111");

        connectionFactory.setPort(5672);

        connectionFactory.setVirtualHost("/");

        connectionFactory.setUsername("drs");

        connectionFactory.setPassword("123456");

        return connectionFactory;

    }

}

码农公寓

1、RPC简述

2、举例说明RabbitMQ中的RPC实现

相关文章