框架功能 - 使用saga - 《Choerodon（猪齿鱼）v0.18 中文文档》

添加依赖
注解
- @SagaTask
Asgard 服务

介绍

Saga 是分布式系统中数据一致性的一种解决方案。本章介绍了如何使用Choerodon 的Saga。并包含如下的内容：

添加依赖
注解
开启Saga
消费者
Asgard 服务

在本地服务的pom.xml 中添加如下的依赖。

消费者端配置

choerodon:
  saga:
    consumer:
      enabled: false # 启动消费端
      thread-num: 2 # saga消息消费线程池大小
      max-poll-size: 200 # 每次拉取消息最大数量
      poll-interval-ms: 1000 # 拉取间隔，默认1000毫秒

在choerodon-starter-asgard 中，包含有两个注解。

@Saga：定义一个类似于Kafka的topic。任务会被添加@SagaTask注解订阅。
@SagaTask：订阅对应的@Saga 的任务。首先，请确保注解所在类可以被spring扫描到

在方法或者类上添加@Saga 注解。

@Saga(code = "asgard-create-user", description = "创建用户", inputSchemaClass = AsgardUser.class)

@SagaTask

在方法上添加@SagaTask注解，@SagaTask本身封装了事务，无需再使用@Transacional声明事务

在一个Saga 定义中。上一个SagaTask的输出是下一个的输入，当seq相同时，则并行执行，并行的任务输出的结果json进行一个合并，作为下一个次序的输入。

并发策略，默认为NONE，TYPE根据sagaClient.startSaga时的refType设置并发，TYPE_AND_ID根据refType和设置并发，并发数为concurrentLimitNum。一个服务将@SagaTask注解删除，asgard服务也会同步删除该SagaTask。

开启Saga

当Saga 被定义好之后，可以通过服务自身，启动一个Saga 实例。

通过Feign启动一个Saga（过时）

注入SagaClient，通过feign调用saga。
将业务代码和sagaClient.startSaga()放在一个事务中。
当不需要消费端消费该Saga实例时，添加choerodon.saga.consumer.enabled: false配置，这样不会创建消费端拉取消息和消息消费的bean和线程。
feign字段
- sagaCode: 要启用的saga的code字段，对应@Saga里的code
- StartInstanceDTO: DTO
  - userId: 方便追踪用户。DetailsHelper.getUserDetails().getUserId()传入
  - refType: 关联业务类型，比如project，user这些。非必须，该字段用于并发策略
  - refId: 关联业务类型，比如projectId，userId这些。非必须，该字段用于并发策略
  - level: 层级。取值”site”、”organization”、”project”、”user”。
  - sourceId: 资源id。比如项目层就为项目id，设置了level和sourceId之后，就可以对应的层级下看到事务实例信息，否则默认全局层，只能在全局层页面看到该事务运行实例信息。请确保@EnableFeignClients包含io.choerodon.asgard.saga，否则扫描不到该feignClient。例如：@EnableFeignClients("io.choerodon")

例如创建一个用户时，启动一个Saga:

@Transactional
    public AsgardUser createUser(@Valid @RequestBody AsgardUser user) {
         // 业务代码
         sagaClient.startSaga("asgard-create-user", new StartInstanceDTO(input, "", ""));
    }

通过TransactionalProducer启动一个Saga

注入TransactionalProducer，通过StartSagaBuilder启动一个Saga
将业务代码和producer.applyAndReturn()放在一个事务中。
当不需要消费端消费该Saga实例时，添加choerodon.saga.consumer.enabled: false配置，这样不会创建消费端拉取消息和消息消费的bean和线程。
方法
- withSagaCode(String sagaCode): 传入要启用的saga的code字段，对应@Saga里的code
- withLevel(ResourceLevel level): 传入层级。取值”site”、”organization”、”project”、”user”
- withRefType(String refType): 传入关联业务类型，该字段用于并发策略
- withRefId(String refId): 传入关联业务Id。非必须，该字段用于并发策略
- withPayloadAndSerialize(Object payload): 传入输入对象例如创建一个用户时，启动一个Saga：

producer.applyAndReturn(
                StartSagaBuilder
                        .newBuilder()
                        .withLevel(ResourceLevel.ORGANIZATION)
                        .withRefType("organization")
                        .withSagaCode("asgard-create-user"),
                builder -> {
                    asgardService.createuser(user); 
                    builder
                            .withPayloadAndSerialize(sagaPayload)
                            .withRefId(String.valueOf(orgId))
                            .withSourceId(orgId);
                    return sagaPayload;
                });

方法返回值为该任务的输出，本次sagaTask的输出是下一个sagaTask的输入。

里面执行封装了事务，不需要再加事务，如果需要加外部事务，可通过@SagaTask的设置事务传播行为。

输出合并

同一个Saga下的多个SagaTask的seq相同，则并行执行。这多个SagaTask的输出进行合并后，成为下个SagaTask的输入。

合并操作如下：Saga1 的code 为code1，Saga2 的code 为code2，如果输出结果完全相同，则合并结果为1或者2的输出。

如下：如果这次的输出和输入一样，直接将接收数据返回即可。

@SagaTask(code = "test", sagaCode = "iam-create-project", seq = 1)
public String iamCreateUser(String data) {
}

这样默认根据方法返回值即String生成的outputChema是错误的，最好手动指定，即:

@SagaTask(code = "test", sagaCode = "iam-create-project", seq = 1,  outputSchemaClass = AsgardUser.class)
public String iamCreateUser(String data) {
    return data;
}

或者指定正确的返回值

一个定时任务线程定时拉取消息，拉取的消息放到一个线程安全的set里，再由消息消费线程池异步消费，每消费完成(无论成功还是失败)set从中删除，直到set为空再进行下一次拉取消费。

消费端事务

@SagaTask注解的方法封装了事务，有如下事务属性可配置。

@SagaTask的方法执行遇到任何异常都会回滚事务，如果无需回滚，则手动捕获该异常即可，如下：

@SagaTask(code = "book-tour-hotel",
        description = "预定酒店",
        sagaCode = "book-tour-package",
        concurrentLimitNum = 2,
        seq = 5)
public TourDTO bookHotel(String data) throws IOException {
    TourDTO tour = mapper.readValue(data, TourDTO.class);
    TourHotel hotel = new TourHotel();
    hotel.setUserId(tour.getUserId());
    hotel.setTourId(tour.getTourId());
    if (tourHotelMapper.insert(hotel) != 1) {
        throw new CommonException("error.tour.bookHotel");
    }
    tour.setHotelId(hotel.getId());
    //比如该feign做一些清理，成功与否无关紧要，则可以手动捕获该异常。
    try {
        XXXFeign.cleanup(tour.getUserId());
    } catch (Exception e) {
        LOGGER
    }
    return tour;
}

@SagaTask的方法里含有feign调用, 最好能保证feign调用的”幂等性”

在北欧神话中，阿斯加德（古诺斯语：Ásgarðr，英语：Asgard）是神的领域，亦可称作阿萨神域。在Choerodon 中，我们用Asgard。来管理choerodon 中所有的分布式事务。

asgard-service 启动后，会主动拉取@Saga和的注解配置。

不存在则插入
存在则更新
原本存在后来删除注解，SagaTask会删除，Saga不做处理。

为了防止消费端多实例拉取出现消费，对每条消息设置一个实例锁，锁为·sagaCode + taskCode`。

当消息实例锁为空时，消费端拉取该条消息并更新实例锁，更新成功，则拉取可以成功