前置条件

在开始使用Saga之前，要确保服务的choerodon-starters-asgard依赖在0.6.3.RELEASE版本及之上, 推荐最新版0.8.1.RELEASE。同时需要对Saga 有一定的了解。可以参考Choerodon猪齿鱼平台中的微服务数据一致性解决方案。

介绍

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

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

添加依赖

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

    <dependency>
        <groupId>io.choerodon</groupId>
        <artifactId>choerodon-starter-asgard</artifactId>
        <version>${choerodon.starters.version}</version>
    </dependency>

消费者端配置

choerodon:
  saga:
    consumer:
      thread-num: 5 # saga消息消费线程池大小
      max-poll-size: 200 # 每次拉取消息最大数量
      enabled: true # 启动消费端，如果服务中不存在@SagaTask，可设置为false，避免资源浪费
      poll-interval-ms: 1000 # 拉取间隔，默认1000毫秒

注解

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

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

@Saga

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

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

@SagaTask

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

@SagaTask(code = "devopsCreateUser",
    description = "devops创建用户",
    sagaCode = "asgard-create-user",
    concurrentLimitNum = 2,
    concurrentLimitPolicy = SagaDefinition.ConcurrentLimitPolicy.NONE,
    seq = 2)

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

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

开启Saga

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

• 注入SagaClient，通过feign调用saga。
• 将业务代码和sagaClient.startSaga()放在一个事务中。
• 当不需要消费端消费该Saga实例时，添加choerodon.saga.consumer.enabled: false配置，这样不会创建消费端拉取消息和消息消费的bean和线程。
• feign字段
  • sagaCode: 要启用的saga的code字段，对应@Saga里的code
  • StartInstanceDTO: DTO
    • input: 输入的json数据
    • 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, "", ""));
    }

同时在代码中添加如下处理逻辑：

@SagaTask(code = "devopsCreateUser",
        description = "devops创建用户",
        sagaCode = "asgard-create-user",
        concurrentLimitNum = 2,
        concurrentLimitPolicy = SagaDefinition.ConcurrentLimitPolicy.NONE,
        seq = 2)
public DevopsUser devopsCreateUser(String data) throws IOException {
    AsgardUser asgardUser = objectMapper.readValue(data, AsgardUser.class);
    LOGGER.info("===== asgardUser {}", asgardUser);
    DevopsUser devopsUser = new DevopsUser();
    devopsUser.setId(asgardUser.getId());
    devopsUser.setGroup("test");
    LOGGER.info("===== devopsCreateUser {}", devopsUser);
    return devopsUser;
}

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

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

输出合并

同一个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) {
    return data;
}

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

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

或者指定正确的返回值

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

消费端模型

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

消费端事务

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

  字段	作用
  transactionIsolation	事务的隔离级别
  transactionManager	使用的事务管理器

`

• 如果@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调用的”幂等性”

Asgard 服务

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

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

• 不存在则插入

• 存在则更新

• 原本存在后来删除注解，SagaTask会删除，Saga不做处理。

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

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

• 当消息实例锁不为空时，查询消息实例是否为拉取的消费端实例，是则允许拉取不是则不允许拉取该条消息。

更多有关的信息可以从asgard-service获取。