可以自学网站开发,域名与网站建设,企业展厅布展设计,学生作业网站2019 年末#xff0c; RocketMQ 正式发布了 4.6.0 版本#xff0c;增加了“ Request-Reply ”的同步调用的新特性。“ Request-Reply ”这个新特性是由微众银行的开发者们总结实践经验#xff0c;并反馈给社区的。接下来本文会详细介绍此新特性。
“ Request-Reply ”是什么…2019 年末 RocketMQ 正式发布了 4.6.0 版本增加了“ Request-Reply ”的同步调用的新特性。“ Request-Reply ”这个新特性是由微众银行的开发者们总结实践经验并反馈给社区的。接下来本文会详细介绍此新特性。
“ Request-Reply ”是什么 图1.1 “ Request-Reply ”模式
在以往的消息中间件的使用中 Producer 和 Consumer 只负责发送消息和消费消息彼此之间不会通信。而 “ Request-Reply ”模式允许 Producer 发出消息后以同步或者异步的形式等待 Consumer 消费这条消息并返回一个响应消息从而达到类似 RPC 的调用效果。在整个“ Request-Reply ”调用过程中简称RR调用 Producer 首先发出一条消息消息经由 Broker 被 Consumer 获取并消费Consumer 消费完这条消息后会将针对该消息的响应作为另外一条消息发送出来最终回到 Producer 。为了便于描述称此时的 Producer 为请求方发出的消息为“请求消息”Consumer 称为服务方返回的消息称为“响应消息”。
“ Request-Reply ”模式使得 RocketMQ 具备了同步调用的能力拓展了 RocketMQ 的使用场景使其具有更多的应用可能性。开发者可以利用这个特性快速搭建自己的消息服务总线实现 RPC 调用框架由于请求以消息的形式存储在 Broker 便于收集信息做调用链追踪和分析在微服务领域也有着广泛的应用场景。
“ Request-Reply ”的实现逻辑
在 RR 调用中涉及到 Producer、Broker、Consumer 三个角色。
Producer 的实现逻辑 图2.1 producer示意图
1、对请求消息增加对应的标识 Producer 发送请求消息时需要在消息的 Properties 里增加RR调用的标识其中关键的字段有 Correlation_Id、REPLY_TO_CLIENT。Correlation_Id 用来唯一标识一次RR请求通过这个属性来匹配同一个RR调用的请求消息和响应消息。REPLY_TO_CLIENT 用来标识请求消息的发出方其值为 Producer 的 ClientId 。
作为请求方的 Producer 只需增加对应标识到消息中在消息的发送逻辑上与原始 Producer 保持一致。
2、发完请求消息后等待响应消息。 请求方每次执行 Request 之后会创建 RequestResponseFuture 对象并且以 Correlation_Id 作为key记录到 ResponseFutureTable 中。执行 Request 的线程通过 RequestResponseFuture 里定义的 CountDownLatch 实现阻塞。当响应消息回到 Producer 实例时根据响应消息中的 Correlation_Id从ResponseFutureTable 中获取对应地 RequestResponseFuture 激活 CountDownLatch 唤醒阻塞的线程执行对响应消息的处理。
图2.2 RequestResponseFuture结构
Consumer 的实现逻辑 图2.3 consumer示意图
Consumer 只需要在正常消费一条请求消息后创建响应消息并发送出去即可。创建响应消息时必须使用提供的工具类来创建避免丢失 Correlation_Id、REPLY_TO_CLIENT 等标识和关联RR请求的属性。
Broker 的实现逻辑 图2.4 Broker示意图
Broker 对请求消息的处理与原先的处理逻辑一样但是对于响应消息则是采用主动 Push 的形式将消息推给请求方。服务方 Consumer 将响应消息发送到 Reply_topic 上 Broker 收到响应消息后会交由 ReplyMessageProcessor 处理。Processor 会将响应消息落到 CommitLog 中并且根据响应消息中的 REPLY_TO_CLIENT 得到请求方的 ClientId 通过 ClientId 找到对应的 Producer 实例及其 Channel 将响应消息直接推送给它。
所有的响应消息都会发送到 Reply_topic 上这个 Topic 是由 Broker 自动创建的系统 Topic 以“集群名 _REPLY_TOPIC ”的格式命名。Reply topic 用于做路由发现让响应消息能够发回到请求消息来源的那个集群目的是保证响应消息回到的 Broker 是请求方有连接的 Broker 。采用 Broker 主动推送响应消息的目的也是为了保证响应消息能够精准回到发出请求消息的实例上。
“ Request-Reply ”在金融场景下的实践
金融业务要求服务要持续稳定能够提供 7x24 小时稳定可用的服务并且容错能力要足够强对节点故障能够快速屏蔽影响保证成功率快速恢复。因此微众银行根据具体的使用场景增加了应用多活、服务就近、熔断等特性构建了安全可靠的金融级消息总线 DeFiBus 。 图3.1 总线架构图
如图所示 DeFiBus 自上而下分别是总线层、应用层、 DB 层。
总线层有两个非常重要的服务分别是 GNS 和 GSL 。对每个客户会根据客户信息并且按照权重分配到规划好的 DCN 内实现数据层面的分片。GNS 服务是在数据层面进行的分片寻址确定客户所在的 DCN 。在服务层面会将服务部署到不同的区域在调用服务时会先访问 GSL 服务做服务层面的分片寻址确定当前要访问的服务在哪个 DCN 。从数据和服务两个维度做分片由 GNS 和 GSL 做分片寻址最终由总线实现请求到 DCN 的自动路由。
请求从流量入口进来经由 GNS 和 GSL 寻址确定服务所在的 DCN 后总线会将请求自动路由到对应服务所在的 DCN 区域交由应用处理。每个 DCN 内的应用只处理本 DCN 内的请求。应用会访问同 DCN 内预先分配的主 DB DB 层会有一个多副本来提高可靠性。
为了提升服务的可用性和可靠性 DeFiBus 的开发者针对“ Request-Reply ”的使用做了多个方面的优化和改造。
快速失败和重试 图3.2 快速失败和重试示意图
从使用方视角来看业务的超时时间等同于整个完整 RR 调用的超时时间。一次RR调用内部会涉及 2 次消息的发送当 Broker 有故障时可能会出现消息发送超时。因此内部发送消息的超时时间设置会根据业务超时时间自动调整为较小的值为失败重试留足更多的时间。比如业务超时时间为 3s 则设置发送消息的超时为 1s 。通过调整消息发送超时时间来快速发现 Broker 的故障。当发现 Broker 的故障后 Producer 会立即重试另外一个 Broker 并隔离失败的 Broker 。在隔离结束前 Producer 不会再将消息发到隔离的 Broker 上。
熔断机制 图3.3 熔断示意图
熔断机制是指当某个队列消息堆积达到指定阈值后不再往这个队列发送消息使得这个队列对应的服务实例暂时熔断。
为了实现熔断机制队列增加了“队列深度”属性。队列深度指一个队列中堆积在 Broker 上未被 Consumer 拉取的消息量。当 Consumer 发生故障或者处理异常首先触发客户端的流控机制随后拉消息请求会被不断地延迟此时消息会堆积在 Broker 上。当 Broker 发现某个队列堆积的消息量超过阈值会标记队列为熔断。Producer 发送消息时如果目标队列已经熔断则会收到队列熔断的响应码并立即重试发送消息到另外的队列同时将熔断的队列标记为隔离。在隔离解除前 Producer 不会再往隔离的队列发送消息。
隔离机制
队列级别的隔离机制主要用于 Producer 的重试和服务的熔断机制。 图3.4 隔离示意图
当 Broker 故障时 Consumer 拉消息会触发隔离机制。原生 RocketMQ 的 Consumer 实现中由 PullMessageService 单个线程向所有 Broker 发送拉消息请求。当这些 Broker 中有节点故障时 PullMessageService 线程会因为与故障 broker 建立连接或者请求响应变慢导致线程暂时阻塞这会让其它正常 Broker 的消息处理耗时变高甚至超时。因此开发者为拉消息增加了一个备用线程一旦发现拉消息的请求执行时间超过阈值则标记这个 Broker 为隔离对应的所有拉消息的请求转交给备用线程执行保证 PullMessageService 执行的都是正常的 Broker 的请求。通过线程隔离来保证部分 Broker 的故障不会影响 Consumer 实例拉消息的时效。
队列动态扩容/缩容
队列动态扩容和缩容目的是保持队列数和 Consumer 实例数的一致使得负载均衡后每个实例消费的队列数一样。在 Producer 均匀发送的情况下使得 Consumer 实例不会因为分到的队列数量不一样而出现负载不均衡。
扩容/缩容通过动态调整 Topic 配置的 ReadQueueNum 和 WriteQueueNum 来实现。
在扩容时首先增加可读队列个数保证 Consumer 先完成监听再增加可写队列个数使得 Producer 可以往新增加的队列发消息。 图3.5 队列扩容示意图
队列缩容与扩容的过程相反先缩小可写队列个数不再往即将缩掉的队列发消息等到 Consumer 将该队列里的消息全部消费完成后再缩小可读队列个数完成缩容过程。 图3.6 队列缩容示意图
负载均衡过渡
RocketMQ Consumer 在负载均衡结果发生变化时会将老结果直接更新为新结果是一个 A 到 B 跳变的过程。当 Consumer 和 Broker 多的时候不同的 Consumer 在负载均衡时获取到的 Consumer 个数以及队列个数可能出现不一致导致负载均衡结果不一致。当结果不一致时就会出现队列漏听和重复听的问题。对于同步调用场景队列出现漏听会导致漏听队列上的消息处理耗时变高甚至超时导致调用失败。
负载均衡过渡则是将负载均衡结果变化过程增加了一个过渡态在过渡态的时候 Consumer 会继续保留上一次负载均衡的结果直到一个负载均衡周期结束或者感知到新的属主已经监听上这个队列时才释放老的结果。 图3.7 负载均衡过渡示意图
同城应用多活
为了达到高可用和容灾的一些要求服务会部署在至少两个数据中心。当一个数据中心有某个服务全部故障不可用时其他数据中心正常的实例能自动接管这部分流量。在部署的时候请求方和服务方在两个数据中心都会部署当两中心都正常时请求方会依照服务就近的原则将请求发到同 IDC 内跨 IDC 只通过心跳维持连接。服务方订阅时优先监听同 IDC 内的队列。 图3.8 正常情况示意图
当且仅当另外一个 IDC 中没有存活的服务实例时服务方才会跨 IDC 接管其他 IDC 的队列。如图当数据中心 2 的应用 B 实例全部挂掉后部署在数据中心 1 的实例 1 、 2 、3 在负载均衡时首先对同 IDC 内的队列进行分配然后检查发现数据中心 2 有队列但无存活的应用B实例此时会将数据中心2的队列分配给数据中心1的实例实现跨 IDC 的自动接管。 图3.9 应用故障情况示意图 当某一个数据中心的 Broker 全部挂掉之后请求方会跨 IDC 进行发送。如图在数据中心 2 的 Broker 全部故障后应用 A 的实例 4~6 会将请求发送到数据中心 1 根据服务就近原则这部分请求会由数据中心 1 的应用 B 实例 1~3 处理从而保证 Broker 故障后经由数据中心 2 进来的请求也能被正常处理。 图3.10 Broker故障情况示意图
四、结语
本文主要介绍了 RocketMQ 新特性——“ Request-Reply ”模式。此模式下 Producer 在发出消息后会等待 Consumer 消费并返回响应消息达到类似 RPC 调用的效果。“ Request-Reply ”模式让 RocketMQ 具备了同步调用的能力在此基础上开发者可以开发更多新的特性。为了更好的服务于金融场景微众银行又增加了应用多活服务就近熔断等新的特性构建了安全可靠的金融级消息总线 DeFiBus 。目前微众银行已经将大部分成果通过 DeFiBus 开源出来后续在分片和寻址方面也会有更通用的实践总结和成果介绍欢迎各位了解关注
作者信息陈广胜Apache RocketMQ CommitterDeFiBus 创始人微众银行技术专家中间件平台相关产品负责人曾就职于 IBM 和华为负责运营商云和大数据平台建设。
原文链接 本文为云栖社区原创内容未经允许不得转载。
