基于Apache ServiceComb Pack的分布式事务框架SAGA

概述

Apache ServiceComb Pack 是一个微服务应用的数据最终一致性解决方案，是一种基于在2PC和TCC两者之间的框架，简单的说是依赖协调器的TCC方式。

特性

• 高可用：支持高可用的集群模式部署。
• 高可靠：所有的关键事务事件都持久化存储在数据库中。
• 高性能：事务事件是通过高性能gRPC来上报的，且事务的请求和响应消息都是通过Kyro进行序列化和反序列化。
• 低侵入：仅需2-3个注解和编写对应的补偿方法即可引入分布式事务。
• 部署简单：支持通过容器（Docker）进行快速部署和交付。
• 补偿机制灵活：支持前向恢复（重试）及后向恢复（补偿）功能。
• 扩展简单：基于Pack架构很容实现多种协调协议，目前支持TCC、Saga协议，未来还可以添加其他协议支持。

Pack架构图

• alpha充当协调者的角色，主要负责对事务的事件进行持久化存储以及协调子事务的状态，使其得以最终与全局事务的状态保持一致。
• omega是微服务中内嵌的一个agent，负责对网络请求进行拦截并向alpha上报事务事件，并在异常情况下根据alpha下发的指令执行相应的补偿操作。

Omega内部运行机制

omega是微服务中内嵌的一个agent。当服务收到请求时，omega会将其拦截并从中提取请求信息中的全局事务id作为其自身的全局事务id（即Saga事件id），并提取本地事务id作为其父事务id。在预处理阶段，alpha会记录事务开始的事件；在后处理阶段，alpha会记录事务结束的事件。因此，每个成功的子事务都有一一对应的开始及结束事件。

服务间通信流程

在服务生产方，omega会拦截请求中事务相关的id来提取事务的上下文。在服务消费方，omega会在请求中注入事务相关的id来传递事务的上下文。通过服务提供方和服务消费方的这种协作处理，子事务能连接起来形成一个完整的全局事务。

具体处理流程

Saga处理场景是要求相关的子事务提供事务处理函数同时也提供补偿函数。Saga协调器alpha会根据事务的执行情况向omega发送相关的指令，确定是否向前重试或者向后恢复。

成功场景

成功场景下，每个开始的事件都会有对应的结束事件。

异常场景

异常场景下，omega会向alpha上报中断事件，然后alpha会向该全局事务的其它已完成的子事务发送补偿指令，确保最终所有的子事务要么都成功，要么都回滚。

超时场景

超时场景下，已超时的事件会被alpha的定期扫描器检测出来，与此同时，该超时事务对应的全局事务也会被中断。

TCC 具体处理流程

TCC(try-confirm-cancel)与Saga事务处理方式相比多了一个Try方法。事务调用的发起方来根据事务的执行情况协调相关各方进行提交事务或者回滚事务。

成功场景

成功场景下， 每个事务都会有开始和对应的结束事件。

异常场景

异常场景下，事务发起方会向alpha上报异常事件，然后alpha会向该全局事务的其它已完成的子事务发送补偿指令，确保最终所有的子事务要么都成功，要么都回滚。

QuickStart

引入Pack的依赖

<dependency>

<groupId>org.apache.servicecomb.pack</groupId>

<artifactId>omega-spring-starter</artifactId>

<version>${pack.version}</version>

</dependency>

<dependency>

<groupId>org.apache.servicecomb.pack</groupId>

<artifactId>omega-transport-resttemplate</artifactId>

<version>${pack.version}</version>

</dependency>

注意: 请将${saga.version}更改为实际的版本号。

添加Saga的注解及相应的补偿方法

以一个转账应用为例：

1.在应用入口添加 @EnableOmega 的注解来初始化omega的配置并与alpha建立连接。

@SpringBootApplication

@EnableOmega

public class Application {

public static void main(String[] args) {

SpringApplication.run(Application.class, args);

}

}

1. 在全局事务的起点添加 @SagaStart 的注解。

@SagaStart(timeout=10)

public boolean transferMoney(String from, String to, int amount) {

transferOut(from, amount);

transferIn(to, amount);

}

注意: 默认情况下，超时设置需要显式声明才生效。

1. 在子事务处添加 @Compensable 的注解并指明其对应的补偿方法。

import javax.transaction.Transactional;

import org.apache.servicecomb.pack.omega.transaction.annotations.Compensable;

@Compensable(timeout=5, compensationMethod="cancel")

public boolean transferOut(String from, int amount) {

repo.reduceBalanceByUsername(from, amount);

}

public boolean cancel(String from, int amount) {

repo.addBalanceByUsername(from, amount);

}

注意: 实现的服务和补偿必须满足幂等的条件。

注意: 默认情况下，超时设置需要显式声明才生效。

注意: 若全局事务起点与子事务起点重合，需同时声明 @SagaStart 和 @Compensable 的注解。

3、增加 alpha.cluster.address 参数

alpha:

cluster:

address: alpha-server.servicecomb.io:8080

4、对转入服务重复第三步即可。

TCC 支持

在对应的方法中添加TccStart 和 Participate标注 以一个转账应用为例：

1、在全局事务的起点添加 @TccStart 的注解。

import org.apache.servicecomb.pack.omega.context.annotations.TccStart;

@TccStart

public boolean transferMoney(String from, String to, int amount) {

transferOut(from, amount);

transferIn(to, amount);

}

Note: 当前TCC还不支持Timeout

2、在子事务尝试方法处添加 @Participate 的注解并指明其对应的执行以及补偿方法名,

import javax.transaction.Transactional;

import org.apache.servicecomb.pack.omega.transaction.annotations.Participate;

@Participate(confirmMethod = "confirm", cancelMethod = "cancel")

@Transactional

public void transferOut(String from, int amount) {

// check banalance

}

@Transactional

public void confirm(String from, int amount) {

repo.reduceBalanceByUsername(from, amount);

}

@Transactional

public void cancel(String from, int amount) {

repo.addBalanceByUsername(from, amount);

}

注意: confirm和cancel方法的参数列表应该与@Participate方法一样，并且它们必须是幂等。我们强烈建议使用Spring的@Transactional来保证本地事务的一致性。

注意: 若全局事务起点与子事务起点重合，需同时声明 @TccStart 和 @Participate 的注解。

3、增加 alpha.cluster.address 参数

alpha:

cluster:

address: alpha-server.servicecomb.io:8080

4、对转入服务重复第二步即可。

参考：https://github.com/apache/incubator-servicecomb-saga/