RabbitMQ(六):回调队列callback queue、关联标识correlation id、实现简单的RPC系统

RabbitMQ(六):回调队列callback queue、关联标识correlation id、实现简单的RPC系统

博客翻译自:RabbitMQ Tutorials Java版


RabbitMQ(一):Hello World程序

RabbitMQ(二):Work Queues、循环分发、消息确认、持久化、公平分发

RabbitMQ(三):Exchange交换器–fanout

RabbitMQ(四):Exchange交换器–direct

RabbitMQ(五):Exchange交换器–topic

RabbitMQ(六):回调队列callback queue、关联标识correlation id、实现简单的RPC系统

RabbitMQ(七):常用方法说明 与 学习小结


远程过程调用(RPC):

在第二篇博客中,我们学会了如何使用工作队列将耗时的任务分发给多个工作者。但假如我们想调用远程电脑上的一个函数(或方法)并等待函数执行的结果,这时候该怎么办呢?好吧,这是一个不同的故事。这种模式通常称为远程过程调用RPC(Remote Procedure Call)。

在今天的教程中,我们将会使用RabbitMQ来建立一个RPC系统:一个客户端和一个可扩展的RPC服务端。因为我们没有任何现成的耗时任务,我们将会创建一个假的RPC服务,它将返回斐波那契数(Fibonacci numbers)。


客户端接口(Client interface):

为了演示如何使用RPC服务,我们将创建一个简单的客户端类。它负责暴露一个名为call的方法,该方法将发送一个RPC请求并阻塞,直到接收到回答。

FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();
String result = fibonacciRpc.call("4");
System.out.println( "fib(4) is " + result);

回调队列(Callback queue):

使用RabbitMQ来做RPC很容易。客户端发送一个请求消息,服务端以一个响应消息回应。为了可以接收到响应,需要与请求(消息)一起,发送一个回调的队列。我们使用默认的队列(Java独有的):

callbackQueueName = channel.queueDeclare().getQueue();

BasicProperties props = new BasicProperties
                            .Builder()
                            .replyTo(callbackQueueName)
                            .build();

channel.basicPublish("", "rpc_queue", props, message.getBytes());

// ... then code to read a response message from the callback_queue ...

消息属性
AMQP 0-9-1协议预定义了消息的14种属性。大部分属性都很少用到,除了下面的几种:

  • deliveryMode:标记一个消息是持久的(值为2)还是短暂的(2以外的任何值),你可能还记得我们的第二个教程中用到过这个属性。
  • contentType:描述编码的mime-typemime-type of the encoding)。比如最常使用JSON格式,就可以将该属性设置为application/json
  • ③ replyTo:通常用来命名一个回调队列。
  • ④ correlationId:用来关联RPC的响应和请求。

我们需要引入一个新的类:

import com.rabbitmq.client.AMQP.BasicProperties;

关联标识(Correlation Id):

在上面的方法中,我们为每一个RPC请求都创建了一个新的回调队列。这样做显然很低效,但幸好我们有更好的方式:让我们为每一个客户端创建一个回调队列

这样做又引入了一个新的问题,在回调队列中收到响应后不知道到底是属于哪个请求的。这时候,CorrelationId就可以派上用场了。对每一个请求,我们都创建一个唯一性的值作为CorrelationId。之后,当我们从回调队列中收到消息的时候,就可以查找这个属性,基于这一点,我们就可以将一个响应和一个请求进行关联。如果我们看到一个不知道的 CorrelationId值,我们就可以安全地丢弃该消息,因为它不属于我们的请求。

你可能会问,为什么要忽视回调队列中的不知道的消息,而不是直接以一个错误失败(failing with an error)。这是由于服务端可能存在的竞争条件。尽管不会,但这种情况仍有可能发生:RPC服务端在发给我们答案之后就挂掉了,还没来得及为请求发送一个确认信息。如果发生这种情况,重启后的RPC服务端将会重新处理该请求(因为没有给RabbitMQ发送确认消息,RabbitMQ会重新发送消息给RPC服务)。这就是为什么我们要在客户端优雅地处理重复响应,并且理想情况下,RPC服务要是幂等的。


总结:

RabbitMQ(六):回调队列callback queue、关联标识correlation id、实现简单的RPC系统

我们的RPC系统的工作流程如下:

当客户端启动后,它会创建一个异步的独特的回调队列。对于一个RPC请求,客户端将会发送一个配置了两个属性的消息:一个是replyTo属性,设置为这个回调队列;另一个是correlation id属性,每一个请求都会设置为一个具有唯一性的值。这个请求将会发送到rpc_queue队列。

RPC工作者(即图中的server)将会等待rpc_queue队列的请求。当有请求到来时,它就会开始干活(计算斐波那契数)并将结果通过发送消息来返回,该返回消息发送到replyTo指定的队列。

客户端将等待回调队列返回数据。当返回的消息到达时,它将检查correlation id属性。如果该属性值和请求匹配,就将响应返回给程序。


放在一块:

计算斐波那契数的任务如下:

private static int fib(int n) {
    if (n == 0) return 0;
    if (n == 1) return 1;
    return fib(n-1) + fib(n-2);
}

我们定义了斐波那契函数,它假设只会输入正整数(不要期望该函数在输入很大的数的时候可以好好工作,它可能是最慢的递归实现)。

RPC服务RPCServer.java的代码如下:

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class RPCServer {

    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) {
        //创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Connection connection = null;
        try {
            connection = factory.newConnection();
            final Channel channel = connection.createChannel();

            //声明队列
            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

            //一次只从队列中取出一个消息
            channel.basicQos(1);

            System.out.println(" [x] Awaiting RPC requests");

            //监听消息(即RPC请求)
            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();

                    //收到RPC请求后开始处理
                    String response = "";
                    try {
                        String message = new String(body, "UTF-8");
                        int n = Integer.parseInt(message);
                        System.out.println(" [.] fib(" + message + ")");
                        response += fib(n);
                    } catch (RuntimeException e) {
                        System.out.println(" [.] " + e.toString());
                    } finally {
                        //处理完之后,返回响应(即发布消息)
                        System.out.println("[server current time] : " + System.currentTimeMillis());
                        channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));

                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };

            channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

            //loop to prevent reaching finally block
            while (true) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException _ignore) {
                }
            }
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (connection != null)
                try {
                    connection.close();
                } catch (IOException _ignore) {
                }
        }
    }
}

RPC服务的代码很直白:

  • (1)开始先建立连接、通道并声明队列。
  • (2)我们可能会运行多个服务进程,为了负载均衡我们通过设置 prefetchCount =1将任务分发给多个服务进程
  • (3)我们使用了basicConsume来连接队列,并通过一个DefaultConsumer对象提供回调。这个DefaultConsumer对象将进行工作并返回响应。

我们的RPC客户端RPCClient代码如下:

package com.maxwell.rabbitdemo;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class RPCClient {

    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";
    private String replyQueueName;

    //定义一个RPC客户端
    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        connection = factory.newConnection();
        channel = connection.createChannel();

        replyQueueName = channel.queueDeclare().getQueue();
    }

    //真正地请求
    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();

        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

        final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);

        channel.basicConsume(replyQueueName, true, 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("[client current time] : " + System.currentTimeMillis());
                    response.offer(new String(body, "UTF-8"));
                }
            }
        });

        return response.take();
    }

    //关闭连接
    public void close() throws IOException {
        connection.close();
    }

    public static void main(String[] argv) {
        RPCClient fibonacciRpc = null;
        String response = null;
        try {
            //创建一个RPC客户端
            fibonacciRpc = new RPCClient();
            System.out.println(" [x] Requesting fib(30)");
            //RPC客户端发送调用请求,并等待影响,直到接收到
            response = fibonacciRpc.call("30");
            System.out.println(" [.] Got '" + response + "'");
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (fibonacciRpc != null) {
                try {
                    //关闭RPC客户的连接
                    fibonacciRpc.close();
                } catch (IOException _ignore) {
                }
            }
        }
    }
}

客户端代码看起来有一些复杂:

  • (1)建立连接和通道,并声明了一个独特的回调队列。
  • (2)订阅这个回调队列,所以我们可以接收RPC响应。
  • (3)call方法执行RPC请求。在call方法中,我们首先生成一个具有唯一性的correlationId值并存在变量corrId中。我们的DefaultConsumer中的实现方法handleDelivery会使用这个值来获取争取的响应。然后,我们发布了这个请求消息,并设置了replyTocorrelationId这两个属性。好了,现在我们可以坐下来耐心等待响应到来了。
  • (4)由于我们的消费者处理(指handleDelivery方法)是在子线程进行的,因此我们需要在响应到来之前暂停主线程(否则主线程结束了,子线程接收到了影响传给谁啊)。使用BlockingQueue是一种解决方案。在这里我们创建了一个阻塞队列ArrayBlockingQueue并将它的容量设为1,因为我们只需要接受一个响应就可以啦。handleDelivery方法所做的很简单,当有响应来的时候,就检查是不是和correlationId匹配,匹配的话就放到阻塞队列ArrayBlockingQueue中。
  • 同时,主线程正等待影响。
  • (5)最终将影响返回给用户了。

现在,可以动手实验了。首先,执行RPC服务端,让它等待请求的到来。

 [x] Awaiting RPC requests

然后,执行RPC客户端,即RPCClient中的main方法,发起请求:

[x] Requesting fib(30)
[client current time] : 1500474305838
 [.] Got '832040'

可以看到,客户端很快就接受到了请求,回头看RPC服务端的时间:

 [.] fib(30)
[server current time] : 1500474305835

上面这种设计并不是RPC服务端的唯一实现,但是它有以下几个重要的优势:

  • ① 如果RPC服务端很慢,你可以通过运行多个实例就可以实现扩展。
  • ② 在RPC客户端,RPC要求发送和接受一个消息。非同步的方法queueDeclare是必须的。这样,RPC客户端只需要为一个RPC请求只进行一次网络往返。

但我们的代码仍然太简单,并没有处理更复杂但也非常重要的问题,像:

  • ① 如果没有服务端在运行,客户端该怎么办
  • ② 客户端应该为一次RPC设置超时吗
  • ③ 如果服务端发生故障并抛出异常,它还应该返回给客户端吗?
  • ④ 在处理消息前,先通过边界检查、类型判断等手段过滤掉无效的消息等

 

说明:

①与原文略有出入,如有疑问,请参阅原文

②原文均是编译后通过javacp命令直接运行程序,我是在IDE中进行的,相应的操作做了修改。

③添加了客户端和服务端执行时间。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/114604.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 【离散数学】集合论 第四章 函数与集合(2) 特殊函数类(单射、满射、双射及其性质、常/恒等函数、置换/排列)「建议收藏」

    【离散数学】集合论 第四章 函数与集合(2) 特殊函数类(单射、满射、双射及其性质、常/恒等函数、置换/排列)「建议收藏」本文属于「离散数学」系列文章之一。这一系列着重于离散数学的学习和应用。由于内容随时可能发生更新变动,欢迎关注和收藏离散数学系列文章汇总目录一文以作备忘。此外,在本系列学习文章中,为了透彻理解离散数学,本人参考了诸多博客、教程、文档、书籍等资料。以下是本文的不完全参考目录,在后续学习中还会逐渐补充:(国外经典教材)离散数学及其应用第七版DiscreteMathematicsandItsApplications7th,作者是KennethH.Rosen,袁崇义译,机械工业出版社离散.

  • 自学编程的8个坑,你踩了几个?第七个坑87%都踩过!

    自学编程的8个坑,你踩了几个?第七个坑87%都踩过!避免这8个坑,你的学习效率会得到很大的提高

  • 币圈新手入门教程 怎么样投资数字货币[通俗易懂]

    币圈新手入门教程 怎么样投资数字货币[通俗易懂]普通人怎么投资数字货币?最近和几位朋友聊到数字货币,发现很多人虽然都想入场交易,但实际上还不是很了解这个东西,这对于新手来说,是非常危险的!笔者尽管现在主要做的是外汇,但是曾经也炒币有一段时间,今天就以自己的经验,给各位币圈新手做个入门普及。数字货币说到数字货币,先要了解区块链,数字货币实际上是区块链的一个产物,数字货币都是由区块链技术产生的,而区块链技术近几年的快速发展,也推动了数字货币…

  • cardboard应用_cardboard怎么用

    cardboard应用_cardboard怎么用GoogleCardboard虚拟现实眼镜开发初步(一)虚拟现实技术简介不得不说这几年虚拟现实技术逐渐火热,伴随着虚拟现实设备的价格迅速平民化,越来越多的虚拟现实设备来到了我们眼前,也因此虚拟现实方面的开发离我们也越来越近。这几年迅速崛起的Oculus,其成功就在于拉近了虚拟现实与群众的距离,把原本价格高不可攀的虚拟现实设备放到了我们可以触手可及的位置,Oculus的技术开辟了

  • mac idea 2021.5 激活码【在线注册码/序列号/破解码】

    mac idea 2021.5 激活码【在线注册码/序列号/破解码】,https://javaforall.cn/100143.html。详细ieda激活码不妨到全栈程序员必看教程网一起来了解一下吧!

  • docker端口映射无法访问的解决

    docker端口映射无法访问的解决表现systemctlstatusdocker,显示正常,可以pull,push,build宿主机访问外网没问题,可以连上ubuntu的阿里的源运行容器映射的端口在本机无法访问,用curl127.0.0.1:端口,显示:curl:(56)Recvfailure:Connectionresetbypeerdockerbuild的时候,使用apt-getinstallxx,无法访问,哪怕镜像源是国内的阿里之类的.在改为dockerbuild–networkho

    2022年10月17日

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号