Actor事件处理框架介绍

Posted 2022-11-28 Qunar_尤雪萍

Actor事件处理框架介绍

酒店交易系统架构介绍

酒店订单交易架构的一个特点是所有订单业务都是基于事件驱动，统一的订单存储服务order-store在完成每次变更存储之后会将变更信息以消息事件发送出来，各个业务系统通过配置qmq Listener获取订单变更事件，通过具体变更信息匹配实现对各自业务系统流程的一个驱动，基于此模型实现酒店订单架构相关的20+个业务系统的业务流程驱动。
下面两个图分别是酒店的订单存储模型和订单变更的消息模型：

遇到的一些问题

我们业务系统在基于事件基础上的业务开发遇到一些共性问题

维护qmq配置

• 订单存储服务发出统一Subject的事件，20+系统都需要去声明Group配置Listener。

消息异常的处理机制

• 在业务系统接收到消息并进行业务处理过程中针对异常case业务代码中往往都需要进行监控打点和日志记录，以便监控业务运行状态已经查问题；
• 异常情况下的重试机制，传统做法是说我直接让qmq重发消息但是针对我们目前订单架构由于底层存储采用的是乐观锁机制在并发业务处理过程中会频繁出现EditionConflictException,目前系统中的版本冲突大概是4qps，因此如果仅仅是依靠qmq消息重试会使得一些业务由于EditionConflict无法及时处理，通用的解决方案是业务系统先做一个自我重试，依旧失败再让qmq消息重发。
• 当业务处理异常的时候，QMQ会发起重试，消息失效后（默认15分钟），它就会停止重试，这种case下如何去监控。

消息处理使多个业务耦合在一起

• 一个store消息来了之后，业务相互影响。例如一个订单变更的操作，可能会影响星券发放，同时也会开启一次追加支付。但是如果这个时候HMS挂了，星券一直发放不成功（非重要业务），但是追加支付受到影响不能开启。同理其中任何一个业务失败的时候，会导致所有业务集体重试一次。造成了巨大的开销。也需要各个业务非常严格地保证幂等。目前最多的会出现一个消息同事处理10个业务逻辑。

消息处理故障处理

• 当业务发生故障后，qmq消息不能及时消费成功导致大量qmq消息和重试消息在短时间里量可能会非常高，最终造成消息阻塞，对系统造成影响，这种情况下的qmq消息频繁重发是没有意义的。
• 在故障恢复之后，对于最终处理失败的消息业务系统没法自己做消息重跑，只能找TCDev做消息重发。

Actor介绍

基于目前业务系统针对酒店订单变更事件业务处理中遇到的共性问题：
-多业务如何解耦；
-业务定制化的重试机制；
-异常的监控日志；
-故障恢复支持；
我们将这些问题抽离业务单独形成了一套统一的客户端解决方案。

多业务场景的解耦

原来qmq消息处理之所以会出现多业务场景，原因是我们的业务处理粒度为消息维度，但是针对每一个特定业务动作它其实只负责匹配变更消息中的特定变更事件，举个例子，对于订单状态和支付状态同时发生变更的消息，我们的订单主流程引擎关心的是订单状态变更事件，而支付流程引擎关心的是支付状态事件，因此在Actor我们将每个actor注册绑定到特定事件维度而不直接绑定到qmq消息上。

重试机制

通过对多业务的解耦，我们会将每个业务actor实例和对应的消息事件封装为相互独立的执行单元，执行单元相互独立去执行，管理自己的执行状态，当某一个执行单元执行失败他会依据自己的重试策略为自己发送一个特定的延迟qmq消息，而不会造成其他成功业务单元的重新执行。
执行图如下：

故障恢复支持

actor框架支持执行单元的落地持久化到数据库，通过注解形式实现可配置，同时actor也有自己的log文件：actor-history.log 、actor-fail.log ，当需要故障恢复是通过这些持久化就能轻松实现异常义务的重新执行。

监控日志

actor提供了非常详细的监控和日志用于跟踪每个actor实例的执行过程以及最终执行状态(例如最终执行失败监控)，同时为每个业务系统也提供了非常完备的订单消息处理的整体监控指标。

Actor Demo

Spring 配置

<?xml version="1.0" encoding="UTF-8"?>
<beans default-autowire="byName"
       xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd
       http://www.qunar.com/schema/qta/actor
       http://www.qunar.com/schema/qta/actor/actor.xsd"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:actor="http://www.qunar.com/schema/qta/actor">
<actor:engine appName="QTA_ORDER_BIZ"  <!-- 业务系统标识 -->
              loader="actorDataLoader"/> <!-- actorDataLoader用于延迟加载订单数据 -->
</beans>

事件定义绑定

• 通过继承OrderActor接口，指定事件类型完成Actor实例的声明
• 通过@QActor标签，将Actor实例注册到ActorEngine，并且标签也提供了重试策略的可配置
• 通过@Qdiff标签可以将Actor实例绑定到特定订单事件

@QActor(retry_interval = 10*60*1000, retry_step = 2.0, retry_maxCount = 20)
@QDiff(section = "main-order", diffType = DiffType.CREATE)
public class NewOrderHelloWorldActor implements OrderActor<MainOrderData, DiffEvent>

    private final static Logger logger = getLogger(NewOrderHelloWorldActor.class);

    @Override
    public void onMessage(ActorMessage<MainOrderData, NewOrderEvent> message) 
        logger.info("hello actor!");