Seata分布式事务-Raft模式(NEW)全面解析

Seata分布式事务-Raft模式(NEW)全面解析

Seata 是一款开源的分布式事务解决方案，star高达22000+，社区活跃度极高，致力于在微服务架构下提供高性能和简单易用的分布式事务服务.

本篇文章将解析Seata-Server Raft模式的源码,带你领略从调研对比,设计,到具体实现,再到知识沉淀的过程.

分享人：陈健斌（funkye） github id: a364176773
介绍：
同盾科技高级开发工程师 、Seata Committer、Sofa-Jraft & Spring cloud alibaba & Mybatis-Plus (by dynamic-datasource) & Apache-Dubbo & Druid Contributor

Seata-Raft模式作者

注: 特此感谢Seata社区所有参与code review,pr讨论的小伙伴及Sofa-JRaft社区的家纯兄指导.

目录

  1. raft模式是什么?
  2. 为什么需要raft模式?
  3. seata-raft模式是如何设计的.
  4. 设计上的一些问题探讨
  5. 总结；

1.Raft模式是什么？

首先我们要明白什么是raft分布式一致性算法,这里直接摘抄sofa-jraft官网的相关介绍:

RAFT 是一种新型易于理解的分布式一致性复制协议，由斯坦福大学的 Diego Ongaro 和 John Ousterhout 提出，作为 RAMCloud 项目中的中心协调组件。Raft 是一种 Leader-Based 的 Multi-Paxos 变种，相比 Paxos、Zab、View Stamped Replication 等协议提供了更完整更清晰的协议描述，并提供了清晰的节点增删描述。 Raft 作为复制状态机，是分布式系统中最核心最基础的组件，提供命令在多个节点之间有序复制和执行，当多个节点初始状态一致的时候，保证节点之间状态一致。

简而言之Seata的Raft模式就是基于Sofa-Jraft组件实现可保证Seata-Server自身的数据一致性和服务高可用.

2.为什么需要raft模式

看完上述的Seata-Raft模式是什么的定义后,是否就有疑问,难道现在Seata-Server就无法保证一致性和高可用了吗?那么我们下面从一致性和高可用来看看目前Seata-Server是如何做的.

2.1.一致性:

在目前Seata的设计中,Server端的作用就是用来保证事务的二阶段被正确的执行,那么我们应该知道被正确的执行的前提条件是什么?

没错,就是事务记录的正确存储和记录,要确保事务记录不丢失,状态不错误下,二阶段驱动所有的Seata-RM时一定是正确的行为,而Seata目前对事务状态和记录是如何存储的?

首先要介绍Seata支持事务存储模式有: file, db, redis 三种模式,根据一致性的排名来db(事务保证)>file(默认异步刷盘)>redis(aof,rdb)

顾名思义:

  • file就是Seata自实现的事务存储,默认以异步形式进行存储事务信息到本地磁盘上,为了兼顾性能,默认是异步化,且事务信息存储在内存中,保证内存和磁盘上的一致性,在Seata-Server(以下简称TC)意外宕机时,会在启动时从磁盘读取事务信息并存储到内存中,以便恢复事务上下文继续运转.
  • db是Seata抽象出的AbstractTransactionStoreManager另一个实现,背后由数据库如pgsql,mysql,oracle来实现事务存储,这个比较好理解,就是对数据库的增删改查,一致性由数据库的本地事务保证,数据也由数据库的保证落盘.
  • redis同上,背后就是利用jedis+lua脚本(部分如竞争锁时)来进行事务的增删改查,数据也是由存储方保证,与db一样在seata中处于计算和存储分离架构设计.

2.2.高可用:

高可用其实简单点理解,就是对集群的容灾,是否在TC宕机后,服务能正常运行,常见的方式就是Provider集群部署(可以说TC就是Seata-Client端的一个Provider),再通过注册中心实时感知TC的上线下,及时的切换到可用的TC.

看起来好像加几台机器,部署即可,但是背后存在了一个问题,如何让多个TC像一个整体一样,即便其中一个宕机后,另一个TC能接手宕机时TC的数据完美善后呢?答案其实很简单,在计算与存储分离的情况下,只要把数据丢到存储方,任意一个TC通过这个公共的存储区域读取即可得到所有TC操作的事务信息,也就保证了高可用的能力.

但是我们也提到了一个前提条件,也就是计算与存储分离下,那么计算与存储一体化设计下为什么不行呢?

这就要说说我们的File实现了,file在一致性那块的描述已经说了,他只是讲数据存储在本地磁盘和TC内存中,这个数据写操作是没有任何同步,也就代表了目前的File模式无法高可用,仅支持单机部署,作为初级的快速入门的简单使用,这也让高性能的file(基于内存)基本被告别生产环境的使用.

3.seata-raft模式是如何设计的

3.1.设计原理

Seata-Raft模式的设计思路是来自于当前无法高可用的file模式的包装,将其的所有增删改操作利用raft算法进行同步,这也就保证了多个TC使用file模式时,数据是一致的,且摒弃了file模式的异步刷盘操作,改为定时刷盘(后面详解)

3.2.流程设计

设计示意如上图,client端在启动时,会通过配置中心读取当前client的事务分组,以及相关的raft集群其中一个节点的ip(当然你不嫌麻烦的话可以把整个集群的节点都写上,参考zookeeper那样),通过将获取元数据的request发向172.17.0.1:8091节点后,tc将会把改default raft集群下的元数据返回,即leader&follower&learner

3.2.1.client初始化源码解析

首先client端的代码比较简单,在远程调用处理的抽象类中,将raft模式进行替换掉现有的负载均衡方式具体代码和类如下:

RaftMetadata

1
2
3
4
5
6
7
8
9
10
11
public class RaftMetadata {
	// leader地址
    private InetSocketAddress leaderAddress;
	// 学习者地址列表
    private List<InetSocketAddress> learners;
	// 跟随者地址列表
    private List<InetSocketAddress> followers;
	// 当前metadata获取时间戳
    private Long timestamp;
.....忽略getset
}

TmNettyRemotingClient

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Override
public void init() {
    // registry processor
    registerProcessor();
    if (initialized.compareAndSet(false, true)) {
        super.init();
        if (io.seata.common.util.StringUtils.isNotBlank(transactionServiceGroup)) {
            initConnection();
        }
    }
}
private void initConnection() {
    getClientChannelManager().reconnect(transactionServiceGroup);
    super.initClusterMetaData();
}
@Override
protected int acquireClusterRetryCount() {
    return ConfigurationFactory.getInstance().getInt(CLIENT_TM_ACQUIRE_CLUSTER_RETRY_COUNT,
            DEFAULT_CLIENT_ACQUIRE_CLUSTER_RETRY_COUNT);
}

RmNettyRemotingClient

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@Override
public void init() {
    // registry processor
    registerProcessor();
    if (initialized.compareAndSet(false, true)) {
        super.init();
        // Found one or more resources that were registered before initialization
        if (resourceManager != null
                && !resourceManager.getManagedResources().isEmpty()
                && StringUtils.isNotBlank(transactionServiceGroup)) {
            initConnection();
        }
    }
}
private void initConnection() {
    getClientChannelManager().reconnect(transactionServiceGroup);
    super.initClusterMetaData();
}
@Override
public int acquireClusterRetryCount() {
    return ConfigurationFactory.getInstance().getInt(CLIENT_ACQUIRE_CLUSTER_RETRY_COUNT,
        DEFAULT_CLIENT_ACQUIRE_CLUSTER_RETRY_COUNT);
}

AbstractNettyRemotingClient

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
// 缓存leader地址
   private final RaftMetadata raftMetadata = new RaftMetadata();
// 当前tc是否是raft集群
   private boolean raftCluster;
// 获取集群元数据失败时的重试次数,由子类实现
   protected int acquireClusterRetryCount = this.acquireClusterRetryCount();

// 初始化获取集群元数据
   protected void initClusterMetaData() {
       raftCluster = acquireClusterMetaData();
       if (raftCluster) {
           findLeaderExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("findLeader", 1, true));
           // The leader election takes 5 second
           findLeaderExecutor.scheduleAtFixedRate(() -> {
               try {
                   acquireClusterMetaData();
               } catch (Exception e) {
                   // prevents an exception from being thrown that causes the thread to break
                   LOGGER.error("failed to get the leader address,error:{}", e.getMessage());
               }
           }, SCHEDULE_INTERVAL_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
       }
   }

// 获取集群元数据
   private boolean acquireClusterMetaData() {
       // 双检锁
       if (raftMetadata.isExpired()) {
           synchronized (raftMetadata) {
               // 双检锁 判断当前是否需要更新元数据
               if (raftMetadata.isExpired()) {
                   // 重试次数
                   for (int i = 0; i < acquireClusterRetryCount; i++) {
                       ClusterMetaDataRequest clusterMetaDataRequest = new ClusterMetaDataRequest();
                       try {
                           // 获取任意一个tc地址
                           String tcAddress = loadBalance(getTransactionServiceGroup(), clusterMetaDataRequest);
                           if (StringUtils.isNotBlank(tcAddress)) {
                               // 获取对应的channel
                               Channel channel = clientChannelManager.acquireChannel(tcAddress);
                               // 生成同步rpc
                               RpcMessage rpcMessage = buildRequestMessage(clusterMetaDataRequest,
                                   ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
                               // 获取response
                               ClusterMetaDataResponse clusterMetaDataResponse =
                                   (ClusterMetaDataResponse)super.sendSync(channel, rpcMessage, 2000);
                               if (clusterMetaDataResponse != null && StringUtils.equalsIgnoreCase(
                                   StoreMode.RAFT.getName(), clusterMetaDataResponse.getMode())) {
                                   if (StringUtils.isNotBlank(clusterMetaDataResponse.getLeaderAddress())) {
                                       // 解析相关信息构建raftMetaData
                                       String[] address = clusterMetaDataResponse.getLeaderAddress().split(":");
                                       InetSocketAddress inetSocketAddress =
                                           new InetSocketAddress(address[0], Integer.parseInt(address[1]));
                                       if (raftMetadata.getLeaderAddress() != null
                                           && !Objects.equals(raftMetadata.getLeaderAddress(), inetSocketAddress)) {
                                           LOGGER.info("seata cluster leader: {}", inetSocketAddress);
                                       }
                                       raftMetadata.setLeaderAddress(inetSocketAddress);
                                       if (StringUtils.isNotBlank(clusterMetaDataResponse.getFollowers())) {
                                           String[] followers =
                                               clusterMetaDataResponse.getFollowers().split(ADDRESS_SPLIT_CHAR);
                                           clientChannelManager.reconnect(Arrays.asList(followers),
                                               getTransactionServiceGroup());
                                           List<InetSocketAddress> list = new ArrayList<>();
                                           for (String follower : followers) {
                                               address = follower.split(ADDRESS_LINK_CHAR);
                                               list.add(
                                                   new InetSocketAddress(address[0], Integer.parseInt(address[1])));
                                           }
                                           raftMetadata.setFollowers(list);
                                       }
                                       if (StringUtils.isNotBlank(clusterMetaDataResponse.getLearners())) {
                                           String[] learners =
                                               clusterMetaDataResponse.getLearners().split(ADDRESS_SPLIT_CHAR);
                                           clientChannelManager.reconnect(Arrays.asList(learners),
                                               getTransactionServiceGroup());
                                           List<InetSocketAddress> list = new ArrayList<>();
                                           for (String learner : learners) {
                                               address = learner.split(ADDRESS_LINK_CHAR);
                                               list.add(
                                                   new InetSocketAddress(address[0], Integer.parseInt(address[1])));
                                           }
                                           raftMetadata.setLearners(list);
                                       }
                                       // 确定是raft模式,定时任务可以启用
                                       return true;
                                   }
                               } else {
                                   // 如果不是raft集群,那么就无需定时刷新leader信息了
                                   break;
                               }
                           }
                       } catch (TimeoutException | FrameworkException e) {
                           LOGGER.error("there is an exception to getting the leader address: {}", e.getMessage(), e);
                       }
                   }
               }
           }
       }
       return false;
   }

// 当leader重选举时,新的leader会通知client修改leader地址
   public void modifyLeader(String ip, int port) {
       synchronized (raftMetadata) {
           InetSocketAddress inetSocketAddress = new InetSocketAddress(ip, port);
           raftMetadata.setLeaderAddress(inetSocketAddress);
           LOGGER.info("seata cluster change leader: {}", inetSocketAddress);
       }
   }

   
   protected String loadBalance(String transactionServiceGroup, Object msg) {
       InetSocketAddress address = null;
       try {
           // 如果raft集群那么需要只拿leader地址进行通信
           if (!(msg instanceof ClusterMetaDataRequest) && raftMetadata != null
               && raftMetadata.getLeaderAddress() != null) {
               address = raftMetadata.getLeaderAddress();
           } else {
               // 非raft模式,或者进行获取元数据可以按照lb算法获取集群中一个tc地址
               @SuppressWarnings("unchecked")
               List<InetSocketAddress> inetSocketAddressList =
                   RegistryFactory.getInstance().aliveLookup(transactionServiceGroup);
               address = this.doSelect(inetSocketAddressList, msg);
           }
       } catch (Exception ex) {
           LOGGER.error(ex.getMessage());
       }
       if (address == null) {
           throw new FrameworkException(NoAvailableService);
       }
       return NetUtil.toStringAddress(address);
   }

3.2.2.server端初始化源码解析

首先我们看SessionHolder这个类,此类保存了TC端的所有sessionmanager的实例,并管理其初始化,在初始化后我们就可以来构建raft的相关服务,以下是源码解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
public static void init(String mode) {
    if (StringUtils.isBlank(mode)) {
        mode = CONFIG.getConfig(ConfigurationKeys.STORE_SESSION_MODE,
                CONFIG.getConfig(ConfigurationKeys.STORE_MODE, SERVER_DEFAULT_STORE_MODE));
    }
    StoreMode storeMode = StoreMode.get(mode);
    if (StoreMode.DB.equals(storeMode)) {
        ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName());
        ASYNC_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName(),
            new Object[]{ASYNC_COMMITTING_SESSION_MANAGER_NAME});
        RETRY_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName(),
            new Object[]{RETRY_COMMITTING_SESSION_MANAGER_NAME});
        RETRY_ROLLBACKING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName(),
            new Object[]{RETRY_ROLLBACKING_SESSION_MANAGER_NAME});
    } else if (StoreMode.RAFT.equals(storeMode) || StoreMode.FILE.equals(storeMode)) {
        String sessionStorePath = CONFIG.getConfig(ConfigurationKeys.STORE_FILE_DIR, DEFAULT_SESSION_STORE_FILE_DIR)
            + separator + System.getProperty(SERVER_SERVICE_PORT_CAMEL);
        if (StringUtils.isBlank(sessionStorePath)) {
            throw new StoreException("the {store.file.dir} is empty.");
        }
        if (StoreMode.RAFT.equals(storeMode)) {
            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.RAFT.getName(),
                    new Object[] {ROOT_SESSION_MANAGER_NAME, sessionStorePath});
        } else {
            ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.FILE.getName(),
                    new Object[] {ROOT_SESSION_MANAGER_NAME, sessionStorePath});
        }
        ASYNC_COMMITTING_SESSION_MANAGER = ROOT_SESSION_MANAGER;
        RETRY_COMMITTING_SESSION_MANAGER = ROOT_SESSION_MANAGER;
        RETRY_ROLLBACKING_SESSION_MANAGER = ROOT_SESSION_MANAGER;
    } else if (StoreMode.REDIS.equals(storeMode)) {
        ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.REDIS.getName());
        ASYNC_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class,
            StoreMode.REDIS.getName(), new Object[]{ASYNC_COMMITTING_SESSION_MANAGER_NAME});
        RETRY_COMMITTING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class,
            StoreMode.REDIS.getName(), new Object[]{RETRY_COMMITTING_SESSION_MANAGER_NAME});
        RETRY_ROLLBACKING_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class,
            StoreMode.REDIS.getName(), new Object[] {RETRY_ROLLBACKING_SESSION_MANAGER_NAME});
    } else {
        // unknown store
        throw new IllegalArgumentException("unknown store mode:" + mode);
    }
    // 尝试初始化raft集群相关配置
    RaftServerFactory.getInstance().init(XID.getIpAddress(),
        Integer.parseInt(System.getProperty(SERVER_RAFT_PORT_CAMEL, "9091")));
    if (RaftServerFactory.getInstance().getRaftServer() != null) {
        // 如果raft模式成功启动,那么将作为其他模式的任务协调者
        DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(StoreMode.RAFT.getName());
    } else {
        DISTRIBUTED_LOCKER = DistributedLockerFactory.getDistributedLocker(storeMode.getName());
    }
    if (RaftServerFactory.getInstance().isRaftMode()) {
        return;
    }
    reload(storeMode);
}

/**
 * Reload.
 *
 * @param storeMode the mode of store
 */
protected static void reload(StoreMode storeMode) {
    if (ROOT_SESSION_MANAGER instanceof Reloadable) {
        ((Reloadable)ROOT_SESSION_MANAGER).reload();
        reload(ROOT_SESSION_MANAGER.allSessions(), storeMode);
    }
}

既然我们看到了在RaftServerFactory中进行了初始化,那么接下来就带大家来继续解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
public void init(String host, int port) {
    // 获取raft集群列表,如果未设置,那么直接不进行初始化
    String initConfStr = CONFIG.getConfig(ConfigurationKeys.SERVER_RAFT_CLUSTER);
    if (StringUtils.isBlank(initConfStr)) {
        if (LOGGER.isWarnEnabled()) {
            LOGGER.warn("initialize SofaJRaft fail , server.raft.cluster is null");
        }
        return;
    }
    final Configuration initConf = new Configuration();
    if (!initConf.parse(initConfStr)) {
        throw new IllegalArgumentException("fail to parse initConf:" + initConfStr);
    }
    // analytic parameter
    final PeerId serverId = new PeerId();
    String serverIdStr = host + ":" + port;
    if (!serverId.parse(serverIdStr)) {
        throw new IllegalArgumentException("fail to parse serverId:" + serverIdStr);
    }
    // 获取当前的store mode
    String mode = CONFIG.getConfig(ConfigurationKeys.STORE_MODE);
    StoreMode storeMode = StoreMode.get(mode);
    if (storeMode.equals(StoreMode.RAFT)) {
        raftMode = true;
    }
    final String dataPath = CONFIG.getConfig(ConfigurationKeys.STORE_FILE_DIR, DEFAULT_SESSION_STORE_FILE_DIR)
        + separator + serverIdStr.split(":")[1];
    final NodeOptions nodeOptions = initNodeOptions(initConf);
    try {
        raftServer = new RaftServer(dataPath, SEATA_RAFT_GROUP, serverId, nodeOptions);
    } catch (IOException e) {
        throw new IllegalArgumentException("fail init raft cluster:" + e.getMessage());
    }
    // 获取状态机
    stateMachine = raftServer.getRaftStateMachine();
    LOGGER.info("started counter server at port:{}", raftServer.getNode().getNodeId().getPeerId().getPort());
    // 如果开启自动加入raft集群,那么启动新的节点时会自动添加该节点至raft集群中,等于自动扩容
    if (CONFIG.getBoolean(SERVER_RAFT_AUTO_JOIN, false)) {
        List<PeerId> currentPeers = null;
        try {
            currentPeers = getCliServiceInstance().getPeers(SEATA_RAFT_GROUP, initConf);
        } catch (Exception e) {
            // In the first deployment, the leader cannot be found
        }
        if (CollectionUtils.isNotEmpty(currentPeers)) {
            if (!currentPeers.contains(serverId)) {
                Status status = getCliServiceInstance().addPeer(SEATA_RAFT_GROUP, initConf, serverId);
                if (!status.isOk()) {
                    LOGGER.error("failed to join the RAFT cluster: {}. Please check the status of the cluster",
                        initConfStr);
                }
            }
        }
    }
}

接着我们看看RaftServer的具体实现类的初始化都做了哪些事RaftServerImpl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
private final Logger logger = LoggerFactory.getLogger(getClass());
private final RaftStateMachine raftStateMachine;
private final RaftGroupService raftGroupService;
private final Node node;
private final CliService cliService;

public RaftServer(final String dataPath, final String groupId, final PeerId serverId, final NodeOptions nodeOptions)
    throws IOException {
    // Initialization path
    FileUtils.forceMkdir(new File(dataPath));

    // Here you have raft RPC and business RPC using the same RPC server, and you can usually do this separately
    final RpcServer rpcServer = RaftRpcServerFactory.createRaftRpcServer(serverId.getEndpoint());
    // Initialize the state machine
    this.raftStateMachine = new RaftStateMachine();
    // Set the state machine to startup parameters
    nodeOptions.setFsm(this.raftStateMachine);
    // Set the storage path
    // Log, must
    nodeOptions.setLogUri(dataPath + File.separator + "log");
    // Meta information, must
    nodeOptions.setRaftMetaUri(dataPath + File.separator + "raft_meta");
    // Snapshot, optional, is generally recommended
    nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
    boolean reporterEnabled = ConfigurationFactory.getInstance().getBoolean(SERVER_RAFT_REPORTER_ENABLED, false);
    nodeOptions.setEnableMetrics(reporterEnabled);
    // Initialize the raft Group service framework
    this.raftGroupService = new RaftGroupService(groupId, serverId, nodeOptions, rpcServer);
    this.cliService = RaftServerFactory.getCliServiceInstance();
    ConfigurationCache.addConfigListener(SERVER_RAFT_CLUSTER, this);
    // 启动raft节点
    this.node = this.raftGroupService.start();
    if (reporterEnabled) {
        final Slf4jReporter reporter = Slf4jReporter.forRegistry(node.getNodeMetrics().getMetricRegistry())
            .outputTo(logger).convertRatesTo(TimeUnit.SECONDS)
            .convertDurationsTo(TimeUnit.MILLISECONDS).build();
        reporter.start(ConfigurationFactory.getInstance().getInt(SERVER_RAFT_REPORTER_INITIAL_DELAY, 30),
            TimeUnit.MINUTES);
    }
}
@Override
public void onChangeEvent(ConfigurationChangeEvent event) {
    if (SERVER_RAFT_CLUSTER.equals(event.getDataId())) {
        final Configuration newConf = new Configuration();
        if (newConf.parse(event.getNewValue())) {
            Node node = getNode();
            // 只有leader可以更改集群信息
            if (node != null && node.isLeader()) {
                CliService cliService = getCliService();
                // 当集群发生改变,自动更改集群信息,可能会触发重选举
                cliService.changePeers(SEATA_RAFT_GROUP, getNode().getOptions().getInitialConf(), newConf);
            }
        }
    }
}

ok,直此TC端的raft初始化相关内容已经解析完毕.

3.2.3.server端raft存储模式源码解析

首先通过上述的server端初始化源码,已经可以初步知晓raft存储模式就是包装了file存储模式,并摒弃他原本的事务存储方式,增加raft的日志和快照存储及数据同步,那么我们现在就来看看相关的类吧

RaftLockManager-Raft的全局锁实现,再老方案中一味的只考虑到follower因与leader执行相同的动作得到相同的结果,故忘记了动作可能存在无意义性,故在新版设计中,将raft的lockmanager直接继承file lock manager即可,并只需要在解锁时进行同步即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@LoadLevel(name = "raft")
public class RaftLockManager extends FileLockManager {
    @Override
    public boolean releaseGlobalSessionLock(GlobalSession globalSession) throws TransactionException {
        GlobalTransactionDO globalTransactionDO = new GlobalTransactionDO();
        globalTransactionDO.setXid(globalSession.getXid());
        RaftSessionSyncMsg raftSyncMsg = new RaftSessionSyncMsg(RELEASE_GLOBAL_SESSION_LOCK, globalTransactionDO);
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        Closure closure = status -> {
            if (status.isOk()) {
                try {
                    completableFuture.complete(super.releaseGlobalSessionLock(globalSession));
                } catch (TransactionException e) {
                    completableFuture.completeExceptionally(e);
                }
            } else {
                completableFuture.completeExceptionally(new TransactionException(TransactionExceptionCode.NotRaftLeader,
                    "seata raft state machine exception: " + status.getErrorMsg()));
            }
        };
        return RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public boolean releaseLock(BranchSession branchSession) throws TransactionException {
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        BranchTransactionDO branchTransactionDO = new BranchTransactionDO();
        branchTransactionDO.setBranchId(branchSession.getBranchId());
        branchTransactionDO.setXid(branchSession.getXid());
        RaftSessionSyncMsg raftSyncMsg = new RaftSessionSyncMsg(RELEASE_BRANCH_SESSION_LOCK, branchTransactionDO);
        Closure closure = status -> {
            if (status.isOk()) {
                try {
                    // ensure consistency through state machine reading
                    completableFuture.complete(super.releaseLock(branchSession));
                } catch (TransactionException e) {
                    completableFuture.completeExceptionally(e);
                }
            } else {
                completableFuture.completeExceptionally(new TransactionException(TransactionExceptionCode.NotRaftLeader,
                    "seata raft state machine exception: " + status.getErrorMsg()));
            }
        };
        return RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    // follower使用本地解锁
    public boolean localReleaseGlobalSessionLock(GlobalSession globalSession) throws TransactionException {
        return super.releaseGlobalSessionLock(globalSession);
    }
    // follower使用本地解锁
    public boolean localReleaseLock(BranchSession branchSession) throws TransactionException {
        return super.releaseLock(branchSession);
    }
}

这样lock的实现就很简单了,无需进行同步,而是在添加分支事务时同步分支时一并处理即可,再后续讲到状态机章节会再次提及

而session这块的实现就是将filesessionmanager进行一个继承,使globalsession和branchsession的增删改行为以raft算法进行同步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@LoadLevel(name = "raft", scope = Scope.PROTOTYPE)
public class RaftSessionManager extends FileSessionManager {

    public RaftSessionManager(String name, String sessionStoreFilePath) throws IOException {
        super(name, sessionStoreFilePath);
        // raft不需要file模式的写磁盘,有raft log和snapshot足以
        write = false;
    }

    @Override
    public void addGlobalSession(GlobalSession globalSession) throws TransactionException {
        super.addGlobalSession(globalSession);
    }

    @Override
    public GlobalSession findGlobalSession(String xid) {
        return super.findGlobalSession(xid);
    }

    @Override
    public void onBegin(GlobalSession globalSession) throws TransactionException {
        // 事务begin时,leader先添加数据,由于是不刷盘如果运行到此行leader出现故障,如宕机,那么此动作未记录raft-log,所以不存在一致性问题
        // 重启时会根据raft log回放,再通过新leader的log进行响应动作处理,达到与leader一致
        // 为什么会选择不在状态机里做事务的添加呢?主要考虑到状态机是一个有序单线程,为了避免阻塞带来的损耗,此行为调整加快了状态机的执行,提升了leader的吞吐
        super.addGlobalSession(globalSession);
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        // 构建raft同步行为
        Closure closure = status -> {
            if (status.isOk()) {
                // leader直接返回true即可
                completableFuture.complete(true);
            } else {
                try {
                    completableFuture.completeExceptionally(
                        new TransactionException(TransactionExceptionCode.NotRaftLeader,
                            "seata raft state machine exception: " + status.getErrorMsg()));
                } finally {
                    try {
                        // 如果此节点出现异常,如leader被重选举为follower,那么直接删除刚才的add事务,这样就回到了一致状态
                        super.removeGlobalSession(globalSession);
                    } catch (TransactionException e) {
                        completableFuture.completeExceptionally(e);
                    }
                }
            }
        };
        GlobalTransactionDO globalTransactionDO = SessionConverter.convertGlobalTransactionDO(globalSession);
        // 创建同步任务消息
        RaftSessionSyncMsg raftSyncMsg = new RaftSessionSyncMsg(ADD_GLOBAL_SESSION, globalTransactionDO);
        // 创建任务
        RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public void onStatusChange(GlobalSession globalSession, GlobalStatus globalStatus) throws TransactionException {
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        Closure closure = closureStatus -> {
            if (closureStatus.isOk()) {
                completableFuture.complete(true);
            } else {
                completableFuture.completeExceptionally(
                    new TransactionException(TransactionExceptionCode.NotRaftLeader,
                        "seata raft state machine exception: " + closureStatus.getErrorMsg()));
            }
        };
        GlobalTransactionDO globalTransactionDO = new GlobalTransactionDO(globalSession.getXid());
        RaftSessionSyncMsg raftSyncMsg =
            new RaftSessionSyncMsg(UPDATE_GLOBAL_SESSION_STATUS, globalTransactionDO, globalStatus);
        RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public void onBranchStatusChange(GlobalSession globalSession, BranchSession branchSession,
        BranchStatus branchStatus) throws TransactionException {
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        Closure closure = closureStatus -> {
            if (closureStatus.isOk()) {
                completableFuture.complete(true);
            } else {
                completableFuture.completeExceptionally(
                    new TransactionException(TransactionExceptionCode.NotRaftLeader,
                        "seata raft state machine exception: " + closureStatus.getErrorMsg()));
            }
        };
        BranchTransactionDO branchTransactionDO = new BranchTransactionDO(globalSession.getXid(), branchSession.getBranchId());
        RaftSessionSyncMsg raftSyncMsg =
                new RaftSessionSyncMsg(UPDATE_BRANCH_SESSION_STATUS, branchTransactionDO, branchStatus);
        RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException {
        // 与添加globalsession一样的目的
        globalSession.add(branchSession);
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        branchSession.setStatus(BranchStatus.Registered);
        Closure closure = status -> {
            if (status.isOk()) {
                completableFuture.complete(true);
            } else {
                try {
                    completableFuture.completeExceptionally(
                        new TransactionException(TransactionExceptionCode.NotRaftLeader,
                            "seata raft state machine exception: " + status.getErrorMsg()));
                } finally {
                    try {
                        globalSession.removeBranch(branchSession);
                    } catch (TransactionException e) {
                        completableFuture.completeExceptionally(e);
                    }
                }
            }
        };
        BranchTransactionDO branchTransactionDO = SessionConverter.convertBranchTransactionDO(branchSession);
        RaftSessionSyncMsg raftSyncMsg = new RaftSessionSyncMsg(ADD_BRANCH_SESSION, branchTransactionDO);
        RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public void onRemoveBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException {
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        Closure closure = closureStatus -> {
            if (closureStatus.isOk()) {
                completableFuture.complete(globalSession.remove(branchSession));
            } else {
                completableFuture.completeExceptionally(
                    new TransactionException(TransactionExceptionCode.NotRaftLeader,
                        "seata raft state machine exception: " + closureStatus.getErrorMsg()));
            }
        };
        BranchTransactionDO branchTransactionDO =
            new BranchTransactionDO(globalSession.getXid(), branchSession.getBranchId());
        RaftSessionSyncMsg raftSyncMsg = new RaftSessionSyncMsg(REMOVE_BRANCH_SESSION, branchTransactionDO);
        RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public void onSuccessEnd(GlobalSession globalSession) throws TransactionException {
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
        Closure closure = status -> {
            if (status.isOk()) {
                try {
                    super.removeGlobalSession(globalSession);
                    completableFuture.complete(true);
                } catch (TransactionException e) {
                    completableFuture.completeExceptionally(e);
                }
            } else {
                completableFuture.completeExceptionally(
                    new TransactionException(TransactionExceptionCode.NotRaftLeader,
                        "seata raft state machine exception: " + status.getErrorMsg()));
            }
        };
        GlobalTransactionDO globalTransactionDO = new GlobalTransactionDO(globalSession.getXid());
        RaftSessionSyncMsg raftSyncMsg = new RaftSessionSyncMsg(REMOVE_GLOBAL_SESSION, globalTransactionDO);
        RaftTaskUtil.createTask(closure, raftSyncMsg, completableFuture);
    }

    @Override
    public void onFailEnd(GlobalSession globalSession) throws TransactionException {
        super.onFailEnd(globalSession);
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public class RaftTaskUtil {

    public static boolean createTask(Closure done, RaftSessionSyncMsg data, CompletableFuture<Boolean> completableFuture)
        throws TransactionException {
        // 创建任务
        final Task task = new Task();
        if (data != null) {
            RaftSyncMsg raftSyncMsg = new RaftSyncMsg();
            raftSyncMsg.setBody(data);
            try {
                task.setData(ByteBuffer.wrap(RaftSyncMsgSerializer.encode(raftSyncMsg)));
            } catch (IOException e) {
                throw new TransactionException(e);
            }
        }
        task.setDone(done == null ? status -> {
        } : done);
        // 执行同步任务,背后是raft一致性算法实现
        RaftServerFactory.getInstance().getRaftServer().getNode().apply(task);
        if (completableFuture != null) {
            // 阻塞等待状态机执行完毕
            return futureGet(completableFuture);
        }
        return true;
    }

    public static boolean createTask(Closure done, CompletableFuture<Boolean> completableFuture)
        throws TransactionException {
        return createTask(done, null, completableFuture);
    }

    public static boolean futureGet(CompletableFuture<Boolean> completableFuture) throws TransactionException {
        try {
            return completableFuture.get();
        } catch (InterruptedException e) {
            throw new GlobalTransactionException(TransactionExceptionCode.FailedWriteSession,
                "Fail to store global session: " + e.getMessage());
        } catch (ExecutionException e) {
            if (e.getCause() instanceof TransactionException) {
                throw (TransactionException)e.getCause();
            } else {
                throw new GlobalTransactionException(TransactionExceptionCode.FailedWriteSession,
                    "Fail to store global session: " + e.getMessage());
            }
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public class RaftSyncMsgSerializer {

    private static final Logger LOGGER = LoggerFactory.getLogger(RaftSyncMsgSerializer.class);

    public static byte[] encode(RaftSyncMsg raftSyncMsg) throws IOException {
        try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos)) {
            // 获取raft同步消息所需要的序列化器,目前支持fst,kryo,hessian
            Serializer serializer =
                EnhancedServiceLoader.load(Serializer.class, SerializerType.getByCode(raftSyncMsg.getCodec()).name());
            // 如果消息body不为空,就进行序列化和压缩body
            Optional.ofNullable(raftSyncMsg.getBody()).ifPresent(value -> raftSyncMsg.setBody(
                CompressorFactory.getCompressor(raftSyncMsg.getCompressor()).compress(serializer.serialize(value))));
            oos.writeObject(raftSyncMsg);
            return bos.toByteArray();
        }
    }

    public static RaftSyncMsg decode(byte[] raftSyncMsgByte) {
        try (ByteArrayInputStream bin = new ByteArrayInputStream(raftSyncMsgByte);
            ObjectInputStream ois = new ObjectInputStream(bin)) {
            // 读取消息
            RaftSyncMsg raftSyncMsg = (RaftSyncMsg)ois.readObject();
            // 根据该同步消息当时的序列化方式获取序列化器
            Serializer serializer =
                EnhancedServiceLoader.load(Serializer.class, SerializerType.getByCode(raftSyncMsg.getCodec()).name());
            // 根据该同步消息当时的压缩方式获取压缩器,解压后进行反序列化
            Optional.ofNullable(raftSyncMsg.getBody())
                .ifPresent(value -> raftSyncMsg.setBody(serializer.deserialize(CompressorFactory
                    .getCompressor(raftSyncMsg.getCompressor()).decompress((byte[])raftSyncMsg.getBody()))));
            return raftSyncMsg;
        } catch (ClassNotFoundException | IOException e) {
            LOGGER.info("Failed to read raft synchronization log: {}", e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

}

如上所述,raftsessionmanager做的任务就是把每次的写操作进行同步到follower节点,通过raft一致性协议保证了多节点的数据一致性,那么这个时候我们就应该去了解状态机的设计,因为这才是同步的核心地方

RaftStateMachine-TC的raft状态机类,此处着重讲述seata利用sofa-jraft状态机做了什么,具体可以访问sofa官网

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
public class RaftStateMachine extends StateMachineAdapter {

    private static final Logger LOGGER = LoggerFactory.getLogger(RaftStateMachine.class);

    private String mode;

    private static final KryoSerializerFactory FACTORY = KryoSerializerFactory.getInstance();

    private static final String BRANCH_SESSION_MAP = "branchSessionMap";

    private static final Map<MsgType, RaftMsgExecute> EXECUTES = new HashMap<>();

    /**
     * Leader term
     */
    private final AtomicLong leaderTerm = new AtomicLong(-1);

    public boolean isLeader() {
        return this.leaderTerm.get() > 0;
    }

    public RaftStateMachine() {
        // 获取当前存储模式
        mode = ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.STORE_MODE);
        // 将相应消息的执行器放入EXECUTES
        EXECUTES.put(ADD_GLOBAL_SESSION, new AddGlobalSessionExecute());
        EXECUTES.put(ADD_BRANCH_SESSION, new AddBranchSessionExecute());
        EXECUTES.put(REMOVE_BRANCH_SESSION, new RemoveBranchSessionExecute());
        EXECUTES.put(UPDATE_GLOBAL_SESSION_STATUS, new UpdateGlobalSessionExecute());
        EXECUTES.put(RELEASE_GLOBAL_SESSION_LOCK, new GlobalReleaseLockExecute());
        EXECUTES.put(REMOVE_GLOBAL_SESSION, new RemoveGlobalSessionExecute());
        EXECUTES.put(UPDATE_BRANCH_SESSION_STATUS, new UpdateBranchSessionExecute());
        EXECUTES.put(RELEASE_BRANCH_SESSION_LOCK, new BranchReleaseLockExecute());
    }

    @Override
    public void onApply(Iterator iterator) {
        while (iterator.hasNext()) {
            if (iterator.done() != null) {
                // leader does not need to be serialized, just execute the task directly
                Optional.ofNullable(iterator.done()).ifPresent(done -> done.run(Status.OK()));
            } else {
                ByteBuffer byteBuffer = iterator.getData();
                // if data is empty, it is only a heartbeat event and can be ignored
                if (byteBuffer != null && byteBuffer.hasRemaining()) {
                    // 反序列化后读取同步消息并执行
                    RaftSessionSyncMsg msg =
                        (RaftSessionSyncMsg)RaftSyncMsgSerializer.decode(byteBuffer.array()).getBody();
                    // follower executes the corresponding task
                    onExecuteRaft(msg);
                }
            }
            iterator.next();
        }
    }

    @Override
    public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
        // 不是raft模式无需做快照
        if (!StringUtils.equals(StoreMode.RAFT.getName(), mode)) {
            done.run(Status.OK());
            return;
        }
        // gets a record of the lock and session at the moment
        Map<String, Object> maps = new HashMap<>(2);
        RaftSessionManager raftSessionManager = (RaftSessionManager)SessionHolder.getRootSessionManager();
        Map<String, GlobalSession> sessionMap = raftSessionManager.getSessionMap();
        Integer initialCapacity = sessionMap.size();
        Map<String, byte[]> globalSessionByteMap = new HashMap<>(initialCapacity);
        // each transaction is expected to have two branches
        Map<Long, byte[]> branchSessionByteMap = new HashMap<>(initialCapacity * 2);
        // 转换globalsession和branchsession
        sessionMap.forEach((k, v) -> {
            globalSessionByteMap.put(v.getXid(), v.encode());
            List<BranchSession> branchSessions = Collections.unmodifiableList(v.getBranchSessions());
            branchSessions.forEach(
                branchSession -> branchSessionByteMap.put(branchSession.getBranchId(), branchSession.encode()));
        });
        maps.put(ROOT_SESSION_MANAGER_NAME, globalSessionByteMap);
        maps.put(BRANCH_SESSION_MAP, branchSessionByteMap);
        RaftSnapshot raftSnapshot = new RaftSnapshot();
        raftSnapshot.setBody(maps);
        LOGGER.info("globalSessionMap size :{}, branchSessionMap map size: {}", globalSessionByteMap.size(),
            branchSessionByteMap.size());
        String path = new StringBuilder(writer.getPath()).append(File.separator).append("data").toString();
        // 保存快照至磁盘
        if (RaftSnapshotFile.save(raftSnapshot, path)) {
            if (writer.addFile("data")) {
                done.run(Status.OK());
            } else {
                done.run(new Status(RaftError.EIO, "Fail to add file to writer"));
            }
        } else {
            done.run(new Status(RaftError.EIO, "Fail to save counter snapshot %s", path));
        }
    }

    @Override
    public boolean onSnapshotLoad(final SnapshotReader reader) {
        // 不是raft模式无需读取快照,因为仅做任务调度
        if (!StringUtils.equals(StoreMode.RAFT.getName(), mode)) {
            return true;
        }
        // leader本身数据就是最新的,无需读取快照
        if (isLeader()) {
            if (LOGGER.isWarnEnabled()) {
                LOGGER.warn("Leader is not supposed to load snapshot");
            }
            return false;
        }
        if (reader.getFileMeta("data") == null) {
            LOGGER.error("Fail to find data file in {}", reader.getPath());
            return false;
        }
        String path = new StringBuilder(reader.getPath()).append(File.separator).append("data").toString();
        try {
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("on snapshot load start index: {}", reader.load().getLastIncludedIndex());
            }
            // 读取快照
            Map<String, Object> maps = RaftSnapshotFile.load(path);
            RaftSessionManager raftSessionManager = (RaftSessionManager)SessionHolder.getRootSessionManager();
            Map<String, byte[]> globalSessionByteMap = (Map<String, byte[]>)maps.get(ROOT_SESSION_MANAGER_NAME);
            Map<Long, byte[]> branchSessionByteMap = (Map<Long, byte[]>)maps.get(BRANCH_SESSION_MAP);
            Map<String, GlobalSession> rootSessionMap = raftSessionManager.getSessionMap();
            // 在使用快照前,需要将本地数据清空,因为快照是全量复制,不是增量形式,所以会和本地已有数据冲突,需要清理后直接覆盖
            LockerManagerFactory.getLockManager().cleanAllLocks();
            rootSessionMap.clear();
            if (!globalSessionByteMap.isEmpty()) {
                Map<String, GlobalSession> sessionMap = new HashMap<>();
                globalSessionByteMap.forEach((k, v) -> {
                    GlobalSession session = new GlobalSession();
                    session.decode(v);
                    sessionMap.put(k, session);
                });
                if (CollectionUtils.isNotEmpty(branchSessionByteMap)) {
                    branchSessionByteMap.forEach((k, v) -> {
                        BranchSession branchSession = new BranchSession();
                        branchSession.decode(v);
                        try {
                            // 重新竞争锁,此处是百分百成功的,因为只有竞争锁成功的分支才会有资格在快照中
                            branchSession.lock();
                            sessionMap.get(branchSession.getXid()).add(branchSession);
                        } catch (TransactionException e) {
                            LOGGER.error(e.getMessage());
                        }
                    });
                    sessionMap.values().parallelStream().forEach(globalSession -> {
                        if (GlobalStatus.Rollbacking.equals(globalSession.getStatus())
                            || GlobalStatus.TimeoutRollbacking.equals(globalSession.getStatus())) {
                            // 重新标记lockstatus
                            globalSession.getBranchSessions().parallelStream()
                                    .forEach(branchSession -> branchSession.setLockStatus(LockStatus.Rollbacking));
                        }
                    });
                }
                // 将处理完的sessionmap放入rootsessionmanager中的sessionmap
                rootSessionMap.putAll(sessionMap);
            }
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("on snapshot load end index: {}", reader.load().getLastIncludedIndex());
            }
            return true;
        } catch (final Exception e) {
            LOGGER.error("fail to load snapshot from {}", path);
            return false;
        }

    }

    @Override
    public void onLeaderStart(final long term) {
        // become the leader again,reloading global session
        if (!isLeader() && RaftServerFactory.getInstance().isRaftMode()) {
            LOGGER.info("session map: {} ",SessionHolder.getRootSessionManager().allSessions().size());
            // reload 将当前所有事务重新处理
            SessionHolder.reload(SessionHolder.getRootSessionManager().allSessions(), StoreMode.RAFT, false);
            NettyRemotingServer nettyRemotingServer =
                (NettyRemotingServer)DefaultCoordinator.getInstance().getRemotingServer();
            LeaderNotifyRequest leaderNotifyRequest = new LeaderNotifyRequest();
            leaderNotifyRequest.setAddress(XID.getIpAddressAndPort());
            // 通知client 该节点成为leader
            nettyRemotingServer.sendSyncRequestAll(leaderNotifyRequest);
        }
        this.leaderTerm.set(term);
        super.onLeaderStart(term);
        DefaultCoordinator.getInstance().setPrevent(true);
    }

    @Override
    public void onLeaderStop(final Status status) {
        this.leaderTerm.set(-1);
        super.onLeaderStop(status);
        // 此处会将当前的tc handle设置为拒绝client的事务请求
        DefaultCoordinator.getInstance().setPrevent(false);
    }

    @Override
    public void onStopFollowing(final LeaderChangeContext ctx) {
        super.onStopFollowing(ctx);
    }

    @Override
    public void onStartFollowing(final LeaderChangeContext ctx) {
        super.onStartFollowing(ctx);
        DefaultCoordinator.getInstance().setPrevent(false);
    }

    private void onExecuteRaft(RaftSessionSyncMsg msg) {
        RaftMsgExecute execute = EXECUTES.get(msg.getMsgType());
        if (execute == null) {
            throw new RuntimeException(
                "the state machine does not allow events that cannot be executed, please feedback the information to the Seata community !!! msg: "
                    + msg);
        }
        try {
            // 执行相应任务
            execute.execute(msg);
        } catch (Throwable e) {
            LOGGER.error("Message synchronization failure: {}, msgType: {}", e.getMessage(), msg.getMsgType(), e);
            throw new RuntimeException(e);
        }
    }
}

我们来重点看下执行器做了什么,首先看下AddGlobalSessionExecute

1
2
3
4
5
6
7
8
9
10
11
@Override
public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
    GlobalSession globalSession = SessionConverter.convertGlobalSession(sessionSyncMsg.getGlobalSession());
    globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager());
    // 很简单添加响应的globalsession到sessionmanager即可
    raftSessionManager.addGlobalSession(globalSession);
    if (logger.isDebugEnabled()) {
        logger.debug("addGlobalSession xid: {},status: {}", globalSession.getXid(), globalSession.getStatus());
    }
    return true;
}

AddBranchSessionExecute,此时我们可以解答一下与老版设计对比,为什么不需要同步锁了,因为在seata-server中当分支被添加的时候,也就是走到sessionmanager中的onaddbranch时,只有获取了锁成功的分支才会走到此处,所以当分支被同步的时候,也就意味着锁一定是争抢成功,这样就避免了很多无意义的锁同步行为,将锁同步和分支事务添加合并为一次同步任务即可.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Override
public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
    BranchTransactionDO branchTransactionDO = sessionSyncMsg.getBranchSession();
    GlobalSession globalSession = raftSessionManager.findGlobalSession(branchTransactionDO.getXid());
    BranchSession branchSession = SessionConverter.convertBranchSession(branchTransactionDO);
    // 竞争锁,或许这里应该叫做获取锁,因为是百分百成功的
    branchSession.lock();
    globalSession.add(branchSession);
    if (logger.isDebugEnabled()) {
        logger.debug("addBranch xid: {},branchId: {}", branchTransactionDO.getXid(),
            branchTransactionDO.getBranchId());
    }
    return true;
}

按照顺序,一个正常的事务流程 addglobalsession->addbranchsession->releaseGlobalSessionLock->,updateglobalsession->updatebranchsession(不一定有,成功执行二阶段后会直接删除branch)->branchReleaseLock->removebranchsession->updateglobalsession(更改事务为end状态)->removeglobalsession,所以我们按顺序看响应的执行器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class UpdateGlobalSessionExecute extends AbstractRaftMsgExecute {

    @Override
    public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
        GlobalSession globalSession = raftSessionManager.findGlobalSession(sessionSyncMsg.getGlobalSession().getXid());
        if (globalSession != null) {
        	// 更新状态即可
            globalSession.setStatus(sessionSyncMsg.getGlobalStatus());
            if (logger.isDebugEnabled()) {
                logger.debug("xid: {},status: {}", globalSession.getXid(), globalSession.getStatus());
            }
        }
        return true;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class UpdateBranchSessionExecute extends AbstractRaftMsgExecute {

    @Override
    public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
        GlobalSession globalSession = raftSessionManager.findGlobalSession(sessionSyncMsg.getBranchSession().getXid());
        BranchSession branchSession = globalSession.getBranch(sessionSyncMsg.getBranchSession().getBranchId());
        BranchStatus status = sessionSyncMsg.getBranchStatus();
        branchSession.setStatus(status);
        if (logger.isDebugEnabled()) {
            logger.debug("update branch: {} , status: {}", branchSession.getBranchId(), branchSession.getStatus());
        }
        return true;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public class RemoveBranchSessionExecute extends AbstractRaftMsgExecute {

    @Override
    public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
        GlobalSession globalSession = raftSessionManager.findGlobalSession(sessionSyncMsg.getBranchSession().getXid());
        if (globalSession != null) {
            BranchSession branchSession = globalSession.getBranch(sessionSyncMsg.getBranchSession().getBranchId());
            if (branchSession != null) {
                raftLockManager.localReleaseLock(branchSession);
                globalSession.remove(branchSession);
            }
            if (logger.isDebugEnabled()) {
                logger.debug("removeBranch xid: {},branchId: {}", globalSession.getXid(),
                    sessionSyncMsg.getBranchSession().getBranchId());
            }
        }
        return true;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class RemoveGlobalSessionExecute extends AbstractRaftMsgExecute {
    // 单线程线程池,线程安全
    private static final ThreadPoolExecutor EXECUTOR =
        new ThreadPoolExecutor(1, 1, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(2048),
            new NamedThreadFactory("RemoveGlobalSessionExecute", 1), new ThreadPoolExecutor.CallerRunsPolicy());

    @Override
    public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) {
        // 加快follower状态机执行,将事务的释放锁和解锁丢入异步线程执行,由于线程安全所以也不存在锁冲突
        EXECUTOR.execute(() -> {
            GlobalSession globalSession =
                raftSessionManager.findGlobalSession(sessionSyncMsg.getGlobalSession().getXid());
            if (globalSession != null) {
                try {
                    raftLockManager.localReleaseGlobalSessionLock(globalSession);
                    raftSessionManager.removeGlobalSession(globalSession);
                    if (logger.isDebugEnabled()) {
                        logger.debug("end xid: {}", globalSession.getXid());
                    }
                } catch (TransactionException e) {
                    logger.error("remove global fail error:{}", e.getMessage());
                }
            }
        });
        return true;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class BranchReleaseLockExecute extends AbstractRaftMsgExecute {

    @Override
    public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
        GlobalSession globalSession =
            SessionHolder.getRootSessionManager().findGlobalSession(sessionSyncMsg.getBranchSession().getXid());
        BranchSession branchSession = globalSession.getBranch(sessionSyncMsg.getBranchSession().getBranchId());
        if (branchSession != null) {
            if (logger.isDebugEnabled()) {
                logger.debug("releaseBranchSessionLock xid: {}", globalSession.getXid());
            }
            // 释放branchsession拥有的全局锁
            return raftLockManager.localReleaseLock(branchSession);
        }
        return false;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class GlobalReleaseLockExecute extends AbstractRaftMsgExecute {

    @Override
    public Boolean execute(RaftSessionSyncMsg sessionSyncMsg) throws Throwable {
        GlobalSession globalSession =
            SessionHolder.getRootSessionManager().findGlobalSession(sessionSyncMsg.getGlobalSession().getXid());
        if (globalSession != null) {
            if (logger.isDebugEnabled()) {
                logger.debug("releaseGlobalSessionLock xid: {}", globalSession.getXid());
            }
            // 释放全局锁
            return raftLockManager.localReleaseGlobalSessionLock(globalSession);
        }
        return false;
    }

}

以上Seata-Server(TC)端的相关涉及源码解读已经完毕,在阅读以上代码后,相信大家已经对client-server两端的实现已经有比较清晰的认知了,接下来我们来看看server与client端在raft下的通信相关设计

3.2.4.Client&Server端通信源码解析

接下来我们可以与client相关的初始化做个连线

AbstractTCInboundHandler类中作为tc接收client请求的处理类,所以获取metadata的相关处理也是写在了此处

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@Override
public ClusterMetaDataResponse handle(ClusterMetaDataRequest request, final RpcContext rpcContext) {
    ClusterMetaDataResponse response = new ClusterMetaDataResponse();
    String mode = ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.STORE_MODE);
    response.setMode(mode);
    if (StringUtils.equalsIgnoreCase(StoreMode.RAFT.getName(), mode)) {
        getCluster(response,request.getGroup());
    }
    response.setResultCode(ResultCode.Success);
    return response;
}

protected ClusterMetaDataResponse getCluster(ClusterMetaDataResponse response, String group) {
    // 获取当前的集群信息,如果不存在,当前肯定不是raft集群部署方式
    String currentConf = ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.SERVER_RAFT_CLUSTER);
    if (!StringUtils.isBlank(currentConf)) {
        // 可以发现request中留了一个group口子,在1.5上目前不会使用,当2.0的multi-raft时需要用group切分集群就会使用到该项配置
        String raftGroup = StringUtils.isNotBlank(group) ? group : SEATA_RAFT_GROUP;
        final Configuration currentClusters = new Configuration();
        if (!currentClusters.parse(currentConf)) {
            throw new IllegalArgumentException("fail to parse initConf:" + currentConf);
        }
        RouteTable routeTable = RouteTable.getInstance();
        // 如果集群信息不相同,需要更新路由表数据
        if (!Objects.equals(routeTable.getConfiguration(raftGroup), currentClusters)) {
            routeTable.updateConfiguration(raftGroup, currentClusters);
        }
        try {
            // 刷新leader信息
            routeTable.refreshLeader(RaftServerFactory.getCliClientServiceInstance(), raftGroup, 1000);
            // 从路由表中查找leader
            PeerId leader = routeTable.selectLeader(raftGroup);
            if (leader != null) {
                // 由于raft节点的同步和选举端口是tc端口+1000,故需要转换成事务使用的端口
                response.setLeaderAddress(leader.getIp() + ":" + (leader.getPort() - DEFAULT_RAFT_PORT_INTERVAL));
                Configuration configuration = routeTable.getConfiguration(raftGroup);
                response.setLearners(configuration.getLearners().parallelStream()
                    .map(learner -> learner.getIp() + ":" + (learner.getPort() - DEFAULT_RAFT_PORT_INTERVAL))
                    .collect(Collectors.joining(",")));
                response.setFollowers(configuration.getPeers().parallelStream()
                    .map(follower -> follower.getIp() + ":" + (follower.getPort() - DEFAULT_RAFT_PORT_INTERVAL))
                    .collect(Collectors.joining(",")));
            }
        } catch (Exception e) {
            LOGGER.error("there is an exception to getting the leader address: {}", e.getMessage(), e);
        }
    }
    // 响应到client端即可
    return response;
}

AbstractTCInboundHandler-处理RM/TM客户端请求消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Override
public <T extends AbstractTransactionRequest, S extends AbstractTransactionResponse> void exceptionHandleTemplate(Callback<T, S> callback, T request, S response) {
    try {
        // 当时拒绝时,会抛出当前不是raft集群leader的异常,此时client会重新刷新集群元数据获取leader,一般来说几秒内服务可以恢复正常
        if (!prevent) {
            throw new TransactionException(TransactionExceptionCode.NotRaftLeader,
                    " The current TC is not a leader node, interrupt processing !");
        }
        super.exceptionHandleTemplate(callback,request,response);
    } catch (TransactionException tex) {
        LOGGER.error("Catch TransactionException while do RPC, request: {}", request, tex);
        callback.onTransactionException(request, response, tex);
    }
}

可以看到与原先的rpc设计上区别就是,如果是raft存储模式,那么必须在状态机完全处理完这个请求的所有操作后再做响应,所以大家会看到在createTask的时候利用了completableFuture进行阻塞等待事务相关增删改的结果,这样就可以让client获得响应的时候是一个正确状态,而非异步不确定状态,很多同学不了解raft一致性算法的可能会有疑问,我通过sofa-jraft官网提供的设计图来解释

img

首先我们看到client端的请求通过一致性算法,将操作,比如我们上述所说的添加/修改/删除事务的操作通过日志形式传递,并在超过半数follower应用了该日志后,leader的状态机才会去做处理,如leader是最后才写数据(相对半数follower),而状态机的执行相对client端是异步(状态机是串行,但是跟client请求不是一个线程),也就是对client端的响应必须在写操作都做完了再响应,所以这里的设计利用了状态机是串行有序的,如果server能响应给client,说明先前的增删改操作是同步完成,数据一致的情况,也代表了数据准确落盘(在raft模式下写入内存),那么此时去响应client端是正确行为,如果直接响应,那么由于状态机和业务线程不是同一个,相对业务线程,写操作完全异步化掉了,业务线程响应给client时,未必数据写完了,这个响应的结果可能就有出入.

3.2.5 raft选主解决分布式任务调度解析

DefaultCoordinator#init 该类是异步提交,回滚重试,重试提交,检查事务超时,undolog 防悬挂记录删除的定时任务初始化函数,我们来看下再集成raft之前,是怎么处理的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
 * Init.
 */
public void init() {
    retryRollbacking.scheduleAtFixedRate(
        () -> SessionHolder.distributedLockAndExecute(RETRY_ROLLBACKING, this::handleRetryRollbacking), 0,
        ROLLBACKING_RETRY_PERIOD, TimeUnit.MILLISECONDS);

    retryCommitting.scheduleAtFixedRate(
        () -> SessionHolder.distributedLockAndExecute(RETRY_COMMITTING, this::handleRetryCommitting), 0,
        COMMITTING_RETRY_PERIOD, TimeUnit.MILLISECONDS);

    asyncCommitting.scheduleAtFixedRate(
        () -> SessionHolder.distributedLockAndExecute(ASYNC_COMMITTING, this::handleAsyncCommitting), 0,
        ASYNC_COMMITTING_RETRY_PERIOD, TimeUnit.MILLISECONDS);

    timeoutCheck.scheduleAtFixedRate(
        () -> SessionHolder.distributedLockAndExecute(TX_TIMEOUT_CHECK, this::timeoutCheck), 0,
        TIMEOUT_RETRY_PERIOD, TimeUnit.MILLISECONDS);

    undoLogDelete.scheduleAtFixedRate(
        () -> SessionHolder.distributedLockAndExecute(UNDOLOG_DELETE, this::undoLogDelete),
        UNDO_LOG_DELAY_DELETE_PERIOD, UNDO_LOG_DELETE_PERIOD, TimeUnit.MILLISECONDS);
}

可以看到以上的定时任务,就是这么简单的审计,定时线程来跑任务即可,我们试想一下,在db,redis模式中,因为存储是用的外部存储来保证多tc的资源共享,那么在定时任务的时候,没有做任何排他操作,很可能会出现,同一个事物需要提交补偿的时候,多个tc的定时任务同时发动,同时再数据库读出了同一批需要重试提交的事务,进行下发,那么会带来什么问题呢?

没错,就是幂等性问题(XA与AT及SAGA模式无影响,TCC需要业务自己保证幂等),且多个tc对同一批的事务进行处理也是极大的资源浪费,这是很不可取的.Seata在先前本来想打算引入分布式任务调度组件来解决此问题,而Raft的集成顺带也就解决了此问题,我们来看一下raft下定时任务是如何处理的.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/**
 * Execute the function after get the distribute lock
 *
 * @param key  the distribute lock key
 * @param func the function to be call
 * @return whether the func be call
 */
public static boolean distributedLockAndExecute(String key, NoArgsFunc func) {
    boolean lock = false;
    try {
        if (lock = acquireDistributedLock(key)) {
            func.call();
        }
    } catch (Exception e) {
        LOGGER.info("Exception running function with key = {}", key, e);
    } finally {
        if (lock) {
            try {
                SessionHolder.releaseDistributedLock(key);
            } catch (Exception ex) {
                LOGGER.warn("release distribute lock failure, message = {}", ex.getMessage(), ex);
            }
        }
    }
    return lock;
}

一幕了然,再未来的版本中,TC端预留了对每个模式的分布式锁实现的接口,后续单独的db,redis模式会实现各自的分布式锁机制,使定时任务不再冲突,而raft模式下,对预留的锁接口是如何实现的呢?其实很简单,如果发现当前是leader节点,就等于拿到锁了

我们来看看raft里对相关定时任务的分布式锁是如何做default处理的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@LoadLevel(name = "raft")
public class RaftDistributedLocker implements DistributedLocker {

    protected static final Logger LOGGER = LoggerFactory.getLogger(
            RedisDistributedLocker.class);

    /**
     * Acquire the distributed lock
     *
     * @param distributedLockDO
     * @return
     */
    @Override
    public boolean acquireLock(DistributedLockDO distributedLockDO) {
        return RaftServerFactory.getInstance().isLeader();
    }


    /**
     * Release the distributed lock
     *
     * @param distributedLockDO
     * @return
     */
    @Override
    public boolean releaseLock(DistributedLockDO distributedLockDO) {
        return true;
    }
    
}

然后我们来看一下RaftServerFactory中的isleader又是如何判断,且兼容不用raft进行选举的场景,比如纯db模式

1
2
3
4
5
public Boolean isLeader() {
    // 如果不是raft模式和raftserver是null,那么就返回true
    // 反之,如果状态机存在,那么直接通过状态机读取当前是否是leader即可
    return !isRaftMode() && raftServer == null || (stateMachine != null && stateMachine.isLeader());
}

而如果是db或redis下不用raft又该怎么处理呢?很简单,直接实现相应的acquireLock,不就把default给重写了吗?这就保证了纯db/redis下由对应的store实现分布式锁的设计,而db/redis+raft选举下,直接由leader执行定时任务即可解决集群下冲突的问题.

至此Seata-Server与Sofa-JRfat的结合源码已经解读完毕,接下来我们来讲讲设计上的一些问题并给出个人的理解.

4.设计上的一些问题探讨

1.为什么要同步file模式的数据

答: 首先file不支持高可用的原因就是数据不同步,导致宕机后事务数据无法共享,部分数据可能被脏写,如AT模式需要全局锁来保证隔离性,如果多个tc使用file模式各自为营,那么client端通过LB策略请求的tc截然不同,必定造成数据混乱,从而影响分布式事务的一致性

2.那为什么不直接同步LOCK数据即可?

答: 无法只同步lock数据,其一是一致性问题,如果通过同步lock的话,其余tc在接收到写请求时,lock数据还未同步到位,此时是从何查起呢?不知源头,不知何时同步,不知该不该阻塞等待,所以基本上主流的分布式一致性算法都是由leader来写入数据,因为leader才是真正数据最准确的存在,且也避免了上述所说的不知道什么时候lock数据才同步到位的尴尬局面,所以无法保证分布式事务的隔离性,必须由leader来处理,并不只是同步数据那么简单.

3.那为什么不独立lock到公共存储如db或者redis呢?

答: 未来会支持这种方式,通过lb策略,使begin时路由到的TC地址作为标识,后续的所有rm都会直接请求begin的那台TC,这样事务信息保证了在此raft-group内是一致准确的.这样在tcc xa saga模式下无需全局锁互斥时,可大幅提升吞吐量,而at需要全局锁互斥必定需要一个公共的地方存放锁,在单raft集群模式下直接存储在raft集群中即可,multi-raft下存储在外置存储器上虽然带来了一定吞吐下降,但相对之前的计算与存储完全分离下理论上有可靠的提升.

4.为什么目前由client端主动请求leader节点,而不像zookeeper之类的进行写请求转发呢?

答: 1.性能考虑: 分布式事务的二阶段下,明显带来的就是多次rpc确认事务决议状态,并下发响应结果造成的性能下降,如果直接由server端进行转发,那么其实也存在转发过程中leader的变更,这种由server端转发的方案会带来一定的性能下降,所以目前没有此考虑. 2.中间件职责不同: zookeeper这类的工具并不会像seata那样会有大量的读写请求,即便使用的tps低一些也是能接受.

5.与redis-cluster/sentinel及db模式主备模式比,raft模式的优劣

答:

redis模式对比:

一.cluster跟sentinel一样,存在数据丢失的可能,cluster是一主一备6台redis服务组成,master同步到slave时无法保证强一致性,且cluster适合超大数据量需要数据分片的场景,当前TC的设计在事务结束后就删除,其实是不太适合的.

二.sentinel也是一样无法保证强一致,比如aof的持久化,极大影响redis吞吐,sentinel可以设置让master得到slave写入响应,但是这样一来也就跟raft无太大差距,如raft也需要让follower写入,自身再写,redis又是外部存储多了一次网络io的开销,再吞吐上理论上是比不过raft模式

db模式对比:

一.db模式在seata中目前尚不支持主备切换,首先虽然根据数据库的redolog&binlog二阶段提交来同步到slave,但是说到底master并不会感知slave是否写入了数据,只是关注是否发出了广播,再极限情况下比如master写入了全局锁后宕机,server端对client响应分支注册成功,此时slave接替master后,又来了一个分支进行update操作去争取全局锁,而此时slave还没收到master的全局锁写入广播(可能由于网络问题),所以另一个分支的update操作就被准许,此时前一个事务如果需要回滚会因为后续的update事务没有被全局锁隔离,导致脏写无法回滚的局面,这是不可取的.

二.目前基于db的模式对磁盘和cpu的要求较高,一些用户使用8c16g配置,ssd硬盘可轻松上1200+tps,而使用同等配置下使用机械硬盘可能只有800+左右的tps.究其原因便是seata-server在事务结束后立马就删除造成的数据库频繁的写擦操作,在并发量大的时候,很可能导致大量的页分裂,页合并等行为有严重影响性能的可能性,而raft模式基于对file模式的包装,基于内存的操作,效率理论上远高于db模式,且也能保证数据的一致性.

5.总结

在Seata未来的发展中,性能是至关重要了,我们要对二阶段协议带来的性能下降做尽可能的优化,如存储方面,资源利用率方面,通信方面.这也就代表了我们需要将更多可掌握的拿捏在手,file模式就是一个自实现的事务存储模式,基于此,我们可在自己能力范畴中,以及社区的力量加强对其的优化,把存储性能掌握在自己手中,而外部存储的方案当然也是有好处的,但是对其调优能力因人而异,维护seata组件的相关同学既要维护server端的相关优化指标,也要关注数据库的相关指标,导致有关注点分散,性能问题不好定位的情况.raft模式的到来,代表了类似于redis模式的基于内存的写操作的性能提升,足够的可操控空间,除去了外部的磁盘&网络io开销,达到一个自主可控的阶段.

当然现阶段可能raft模式还不太成熟,未必有上述描述的那般美好,但更因为理论如此,我们应该让事实如此,在此我们欢迎任何对Seata有兴趣的同学加入社区,共同维护,共同进步,一起为Seata分布式事务添砖加瓦!

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码
本站总访问量次 |

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

好记性不如烂笔头,记录敲代码中的点点滴滴.