RbbitMQ之RPC实现（八）

深碍√TFBOYSˉ_ 2022-08-29 10:56 110阅读 0赞

RPC是Remote Procedure Call的简称，即远程过程调用。它是一种通过网络从远程计算机上请求服务（接口调用）。使用RabbitMQ如何实现RPC远程调用呢，服务器消费者订阅一个队列，客户端消费者订阅一个队列，客户端发起请求将消息、请求标识、回调队列发送给服务器订阅的队列，服务器收到消息处理数据将处理结果和客户端的请求标识发送给回调队列。

RPC服务端

ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(AMQP.PROTOCOL.PORT);
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(queueName, false, false, false, null);
            channel.basicQos(1);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body);
                    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
                    builder.correlationId(properties.getCorrelationId());
                    String response = message + "你好RPC";
                    channel.basicPublish("", properties.getReplyTo(), builder.build(), response.getBytes());
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(queueName, false, consumer);

RPC客户端

public String requestQueueName = "rpc_queue";
    
        public String replyQueueName;
    
        public Channel channel;
    
        public DefaultConsumer defaultConsumer;
    
        public String corrId;
    
        public RPCClient() throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(AMQP.PROTOCOL.PORT);
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = factory.newConnection();
            channel = connection.createChannel();
            replyQueueName = channel.queueDeclare().getQueue();
            defaultConsumer = new DefaultConsumer(channel) {
                @Override
                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));
                    }
                }
            };
            channel.basicConsume(replyQueueName, true, defaultConsumer);
        }
    
        public void call(String message) throws Exception {
            corrId = UUID.randomUUID().toString();
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                    .correlationId(corrId)
                    .replyTo(replyQueueName)
                    .build();
            channel.basicPublish("", requestQueueName, properties, message.getBytes());
        }
    
        public static void main(String[] args) throws Exception {
            RPCClient rpcClient = new RPCClient();
            rpcClient.call("哈哈");
        }

注：队列没有绑定交换器的话会绑定默认交换器，bindingKey就是队列的名称。