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

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

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

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

Seata 存储模式Raft&Redis 作者

注: 特此感谢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模式的异步刷盘操作,改为通过raftlog+snapshot恢复数据

3.2.流程设计

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

如图是启动时拉取metadata的过程,及请求过程

假设tm begin一个事务,那么由于本地的metadata指向了tc1的地址,那么他只会与tc1进行交互,而tc1在增加一个全局事务信息时,会通过raft协议,也就是图上标注的步骤1,发送日志,步骤2follower应答日志接收情况,当超过半数节点,如tc2接受成功并相应,那么tc1上的状态机(fsm)将执行添加全局事务动作(详细下面会说到).

那么如果tc1宕机了,或者重选举了会发生什么呢?由于第一次启动拉取到了metadata,client便会执行watch follower节点的watch接口(因为只有非leader才可能成为leader,watch leader是无意义的,且集群发生重选举或增删节点,每个tc节点都会感知)

当tc1宕机后,tc2选举成为leader,而tm通过tc3或者tc2(只要是follower)的watch成功更新本地的metadata信息,后续的事务将请求至tc2,而tc1的数据本身就被同步到了tc2和tc3,数据一致性没影响,只不过在选举发生的一瞬间,且如果事务恰好处于决议发送request至旧leader时,该事务会被主动回滚(因为rpc的节点已经宕机或者重选举了,目前没做该rpc重试,tm侧默认有5次,但由于选举大约需要1s-2s时间,大概率处于这个阶段的事务会回滚)

3.2.1.client初始化源码解析

首先client端通过额外实现一个SeataRegistryServiceImpl为Seata自身作为注册中心

public class SeataRegistryServiceImpl implements RegistryService<ConfigChangeListener> {
    private static final Logger LOGGER = LoggerFactory.getLogger(SeataRegistryServiceImpl.class);

    private static volatile SeataRegistryServiceImpl instance;

    private static final Configuration CONFIG = ConfigurationFactory.getInstance();
	// 代理fileregistry,用于发现Seata Server地址
    private static final RegistryService<?> FILE_REGISTRY_SERVICE = FileRegistryServiceImpl.getInstance();

    private static final String IP_PORT_SPLIT_CHAR = ":";

    private static final String ENDPOINT_AGAIN_SPLIT_CHAR = ",";

    private static final Map<String, List<InetSocketAddress>> INIT_ADDRESSES = new HashMap<>();
	// 缓存元数据
    private static final Metadata METADATA = new Metadata();

    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
	// httpclient 使用
    private static final PoolingHttpClientConnectionManager POOLING_HTTP_CLIENT_CONNECTION_MANAGER =
        new PoolingHttpClientConnectionManager();
	// 当前使用的group
    private static volatile String CURRENT_TRANSACTION_SERVICE_GROUP;
	// 定时刷新元数据的线程池
    private static volatile ThreadPoolExecutor REFRESH_METADATA_EXECUTOR;

    /**
     * Service node health check
     */
    private static final Map<String,List<InetSocketAddress>> ALIVE_NODES = new ConcurrentHashMap<>();

    static {
        // 十个group应该够了吧
        POOLING_HTTP_CLIENT_CONNECTION_MANAGER.setMaxTotal(10);
        POOLING_HTTP_CLIENT_CONNECTION_MANAGER.setDefaultMaxPerRoute(10);
    }

    private SeataRegistryServiceImpl() {
    }

    /**
     * Gets instance.
     *
     * @return the instance
     */
    static SeataRegistryServiceImpl getInstance() {
        if (instance == null) {
            synchronized (SeataRegistryServiceImpl.class) {
                if (instance == null) {
                    instance = new SeataRegistryServiceImpl();
                }
            }
        }
        return instance;
    }

    @Override
    public void register(InetSocketAddress address) throws Exception {

    }

    @Override
    public void unregister(InetSocketAddress address) throws Exception {

    }

    @Override
    public void subscribe(String cluster, ConfigChangeListener listener) throws Exception {

    }

    @Override
    public void unsubscribe(String cluster, ConfigChangeListener listener) throws Exception {

    }

    protected static void startQueryMetadata() {
        if (REFRESH_METADATA_EXECUTOR == null) {
            synchronized (INIT_ADDRESSES) {
                if (REFRESH_METADATA_EXECUTOR == null) {
                    REFRESH_METADATA_EXECUTOR = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                        new LinkedBlockingQueue<>(), new NamedThreadFactory("refreshMetadata", 1, true));
                    REFRESH_METADATA_EXECUTOR.execute(() -> {
                        long metadataMaxAgeMs = CONFIG.getLong(CLIENT_METADATA_MAX_AGE_MS, 30000L);
                        long currentTime = System.currentTimeMillis();
                        while (true) {
                            // 超过指定时间后无论metadata是否有更新,都强制刷新一次
                            boolean fetch = System.currentTimeMillis() - currentTime > metadataMaxAgeMs;
                            if (!fetch) {
                                fetch = watch(currentTime);
                            }
                            // 达到刷新时间或watch接口的response httpstatus为200,代表集群发生变更
                            if (fetch) {
                                // 并行流处理,加快对本地各个group的metadata获取
                                METADATA.groups().parallelStream().forEach(group -> {
                                    try {
                                        acquireClusterMetaData(group);
                                    } 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());
                                    }
                                });
                                currentTime = System.currentTimeMillis();
                            }
                        }
                    });
                    Runtime.getRuntime().addShutdownHook(new Thread(FIND_LEADER_EXECUTOR::shutdown));
                }
            }
        }
    }

    // 获取此次group下http请求的地址
    private static String queryHttpAddress(String group) {
        // 获取group下的node集合
        List<Node> nodeList = METADATA.getNodes(group);
        List<String> addressList = null;
        Stream<InetSocketAddress> stream = null;
        if (CollectionUtils.isNotEmpty(nodeList)) {
            // 获取当前使用的group的存活节点集合
            List<InetSocketAddress> inetSocketAddresses = ALIVE_NODES.get(CURRENT_TRANSACTION_SERVICE_GROUP);
            if (CollectionUtils.isEmpty(inetSocketAddresses)) {
                // 直接转成ip:port
                addressList = nodeList.parallelStream()
                    .map(node -> node.getHostAddress() + IP_PORT_SPLIT_CHAR + node.getHttpPort())
                    .collect(Collectors.toList());
            } else {
                // 获取并行流
                stream = inetSocketAddresses.parallelStream();
            }
        } else {
            // 首次请求时只有本地地址,所以从本地初始化地址获取
            stream = INIT_ADDRESSES.get(group).parallelStream();
        }
        // 如果存活节点列表为null,那么转换metadata或本地地址的port为httpport
        addressList = addressList != null ? addressList
            : stream.map(inetSocketAddress -> inetSocketAddress.getAddress().getHostAddress() + IP_PORT_SPLIT_CHAR
                + (inetSocketAddress.getPort() - 1000)).collect(Collectors.toList());
        int length = addressList.size();
        // 随机获取一个节点
        return addressList.get(ThreadLocalRandom.current().nextInt(length));
    }

    @Override
    public void close() throws Exception {
    }

    // 地址转换
    private InetSocketAddress convertInetSocketAddress(Node node) {
        String ip = node.getHostAddress();
        int port = node.getNettyPort();
        return new InetSocketAddress(ip, port);
    }

    @Override
    public List<InetSocketAddress> lookup(String key) throws Exception {
        String clusterName = getServiceGroup(key);
        if (clusterName == null) {
            return null;
        }
        CURRENT_TRANSACTION_SERVICE_GROUP = key;
        // 本地缓存无此group,说明是刚启动或者切换了事务分组对应的group
        if (!METADATA.containsGroup(clusterName)) {
            List<InetSocketAddress> list = FILE_REGISTRY_SERVICE.lookup(key);
            if (CollectionUtils.isEmpty(list)) {
                return null;
            }
            INIT_ADDRESSES.put(clusterName, list);
            // 首次使用该group必须立马刷新metadata
            acquireClusterMetaData(clusterName);
            // 启动异步watch线程
            startQueryMetadata();
        }
        // 由于上面刷新了metadata,本地就可以直接对reconnect线程给予集群所有节点信息
        List<Node> nodes = METADATA.getNodes(clusterName);
        if (CollectionUtils.isNotEmpty(nodes)) {
            return nodes.parallelStream().map(this::convertInetSocketAddress).collect(Collectors.toList());
        }
        return Collections.emptyList();
    }

    @Override
    public List<InetSocketAddress> aliveLookup(String transactionServiceGroup) {
        // 如果是raft模式,那么需要读取group中的leader进行通信
        if (METADATA.isRaftMode()) {
            String clusterName = getServiceGroup(transactionServiceGroup);
            Node leader = METADATA.getLeader(clusterName);
            if (leader != null) {
                return Collections.singletonList(new InetSocketAddress(leader.getHostAddress(), leader.getNettyPort()));
            }
        }
        // 非raft模式直接读取所有存活节点即可
        return RegistryService.super.aliveLookup(transactionServiceGroup);
    }

    @Override
    public List<InetSocketAddress> refreshAliveLookup(String transactionServiceGroup,
        List<InetSocketAddress> aliveAddress) {
        if (METADATA.isRaftMode()) {
            Node leader = METADATA.getLeader(getServiceGroup(transactionServiceGroup));
            String ip = leader.getHostAddress();
            int port = leader.getNettyPort();
            return ALIVE_NODES.put(transactionServiceGroup,
                aliveAddress.isEmpty() ? aliveAddress : aliveAddress.parallelStream().filter(inetSocketAddress -> {
                    // 只有follower会成为leader所以无需将存活节点中放置leader节点
                    return inetSocketAddress.getPort() != port || !inetSocketAddress.getAddress().getHostAddress().equals(ip);
                }).collect(Collectors.toList()));
        } else {
            // 非raft存储模式直接存放检测过健康状态的集合即可
            return RegistryService.super.refreshAliveLookup(transactionServiceGroup, aliveAddress);
        }
    }

   private static boolean watch(long lastUpdateTime) {
        Map<String, String> param = new HashMap<>();
        StringJoiner stringJoiner = new StringJoiner(ENDPOINT_AGAIN_SPLIT_CHAR);
        // watch当前所用过的所有group
        METADATA.groups().parallelStream().forEach(stringJoiner::add);
        param.put("groupIds", stringJoiner.toString());
        // 提交最后一次更新的时间,如果在30s长轮询恰好结束重新发起期间出现节点更换,那么通过这个时间戳服务端可直接响应而不需要继续挂起30s 
        param.put("lastUpdateTime", String.valueOf(lastUpdateTime));
        String tcAddress = queryHttpAddress(DEFAULT_SEATA_GROUP);
        try (CloseableHttpResponse response = doGet("http://" + tcAddress + "/metadata/v1/watch", param, null, 30000)) {
            if (response != null) {
                StatusLine statusLine = response.getStatusLine();
                // httpstatus 为 200代表集群发生变动,304则为无需刷新元数据
                return statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
            }
        } catch (Exception e) {
            // 如果请求不到可能网络有问题,等待1秒重试
            LOGGER.error("watch cluster fail: {}", e.getMessage());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException ignored) {
            }
        }
        return false;
    }

    
    private static void acquireClusterMetaData(String group) {
        if (METADATA.isExpired(group)) {
            synchronized (group.intern()) {
                if (METADATA.isExpired(group)) {
                    String tcAddress = queryHttpAddress(group);
                    if (StringUtils.isNotBlank(tcAddress)) {
                        Map<String, String> param = new HashMap<>();
                        param.put("group", group);
                        String response = null;
                        try (CloseableHttpResponse httpResponse =
                            doGet("http://" + tcAddress + "/metadata/v1/cluster", param, null, 1000)) {
                            if (httpResponse != null
                                && httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                                response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
                            }
                            MetadataResponse metadataResponse = null;
                            if (StringUtils.isNotBlank(response)) {
                                try {
                                    metadataResponse = OBJECT_MAPPER.readValue(response, MetadataResponse.class);
                                } catch (JsonProcessingException e) {
                                    LOGGER.error(e.getMessage(), e);
                                    return;
                                }
                            }
                            if (metadataResponse != null) {
                                List<Node> list = new ArrayList<>();
                                // 更新元数据
                                for (Node node : metadataResponse.getNodes()) {
                                    if (node.getRole() == ClusterRole.LEADER) {
                                        METADATA.setLeader(node);
                                    }
                                    list.add(node);
                                }
                                METADATA.setStoreMode(StoreMode.get(metadataResponse.getMode()));
                                METADATA.setNodes(group, list);
                            }
                        } catch (IOException e) {
                            LOGGER.error(e.getMessage(), e);
                        }
                    }
                }
            }
        }
    }

    // get请求方法
    public static CloseableHttpResponse doGet(String url, Map<String, String> param, Map<String, String> header,
        int timeout) {
        try {
            URIBuilder builder = new URIBuilder(url);
            if (param != null) {
                for (String key : param.keySet()) {
                    builder.addParameter(key, param.get(key));
                }
            }
            URI uri = builder.build();
            HttpGet httpGet = new HttpGet(uri);
            if (header != null) {
                header.forEach(httpGet::addHeader);
            }
            CloseableHttpClient client =
                HttpClients.custom().setConnectionManager(POOLING_HTTP_CLIENT_CONNECTION_MANAGER)
                    .setDefaultRequestConfig(RequestConfig.custom().setConnectionRequestTimeout(timeout)
                        .setSocketTimeout(timeout).setConnectTimeout(timeout).build())
                    .build();
            return client.execute(httpGet);
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
        return null;
    }

}

通过Seata自身作为client的注册中心后,metadata便成为核心点,具体代码和类如下:

Metadata 其刷新metadata设计线程全为单线程无线程安全问题,且刷新方式为直接put覆盖,思想为copyonwrite,rpc线程获取的时候不受其刷新干扰且无锁

public class Metadata {
	// 存储每个raft-group下的leader
    private final Map<String, Node> leaders = new HashMap<>();
    // 存储每个raft-group的metadata最近一次更新时间
    private final Map<String, Long> leaderUpdateTimes = new HashMap<>();
    // 存储每个raft-group的node的详细信息
    private final Map<String, List<Node>> nodes = new HashMap<>();

    private StoreMode storeMode = StoreMode.FILE;

    public Node getLeader() {
        return getLeader(DEFAULT_SEATA_GROUP);
    }

    public Node getLeader(String group) {
        return leaders.get(group);
    }

    public void setLeaderNode(Node node) {
        String group = node.getGroup();
        this.leaders.put(node.getGroup(), node);
        this.leaderUpdateTimes.put(group, System.currentTimeMillis());
    }

    public boolean isExpired() {
        return isExpired(DEFAULT_SEATA_GROUP);
    }

    public boolean isExpired(String group) {
        Long timestamp = leaderUpdateTimes.get(group);
        return timestamp == null || (System.currentTimeMillis() - timestamp) > 1000;
    }

    public boolean isNotExpired() {
        return !isExpired();
    }

    public void setLeader(Node leader) {
        setLeader(DEFAULT_SEATA_GROUP, leader);
    }

    public void setLeader(String group, Node leader) {
        this.leaders.put(group, leader);
    }

    public List<Node> getNodes() {
        return getNodes(DEFAULT_SEATA_GROUP);
    }

    public List<Node> getNodes(String group) {
        return nodes.get(group);
    }

    public void setNodes(String group, List<Node> nodes) {
        this.nodes.put(group, nodes);
    }

    public void setNodes(List<Node> nodes) {
        setNodes(DEFAULT_SEATA_GROUP, nodes);
    }

    public boolean containsGroup(String group) {
        return nodes.containsKey(group);
    }

    public Set<String> groups() {
        return nodes.keySet();
    }

    public StoreMode getStoreMode() {
        return storeMode;
    }

    public boolean isRaftMode() {
        return Objects.equals(storeMode, StoreMode.RAFT);
    }

    public void setStoreMode(StoreMode storeMode) {
        this.storeMode = storeMode;
    }

    @Override
    public String toString() {
        return StringUtils.toString(this);
    }

}

public class Node {

    private String hostAddress;

    private int nettyPort;
	// 预留未来grpc的端口
    private int grpcPort;
	// http业务侧端口
    private int httpPort;
	// 所属raft-group 默认default
    private String group;
	// db+raft模式下,替代第三方注册中心时,角色都为MEMBER,其余为leader/follower/learner
    private ClusterRole role = ClusterRole.MEMBER;

    public String getHostAddress() {
        return hostAddress;
    }

    public void setHostAddress(String hostAddress) {
        this.hostAddress = hostAddress;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public ClusterRole getRole() {
        return role;
    }

    public void setRole(ClusterRole role) {
        this.role = role;
    }

    public int getNettyPort() {
        return nettyPort;
    }

    public void setNettyPort(int nettyPort) {
        this.nettyPort = nettyPort;
    }

    public int getGrpcPort() {
        return grpcPort;
    }

    public void setGrpcPort(int grpcPort) {
        this.grpcPort = grpcPort;
    }

    public int getHttpPort() {
        return httpPort;
    }

    public void setHttpPort(int httpPort) {
        this.httpPort = httpPort;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        Node node = (Node)o;
        // 去重,因为leader也在成员中
        return Objects.equals(hostAddress, node.hostAddress) && Objects.equals(nettyPort, node.nettyPort)
            && Objects.equals(grpcPort, node.grpcPort) && Objects.equals(httpPort, node.httpPort)
            && Objects.equals(group, node.group);
    }

    @Override
    public int hashCode() {
        return Objects.hash(hostAddress, nettyPort, grpcPort, httpPort, group);
    }

3.2.2.server端初始化源码解析

接下来就是server如何处理watch和刷新元数据接口的请求

@RestController
@RequestMapping("/metadata/v1")
public class ClusterController {

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

	// 观察者管理器
    @Resource
    private ClusterWatcherManager clusterWatcherManager;

    @Resource
    private ApplicationContext applicationContext;

    private ServerProperties serverProperties;

    @PostConstruct
    public void init() {
        serverProperties = applicationContext.getBean(ServerProperties.class);
    }

    @GetMapping("/cluster")
    public MetadataResponse cluster(@RequestParam(defaultValue = DEFAULT_SEATA_GROUP) String group) {
        String mode = ConfigurationFactory.getInstance().getConfig(STORE_MODE);
        MetadataResponse metadataResponse = new MetadataResponse();
        // 填充当前tc的事务模式
        metadataResponse.setMode(mode);
        // 获取对应的raft-group的raftserver
        RaftServer raftServer = RaftServerFactory.getInstance().getRaftServer(group);
        if (raftServer != null) { 
            // 读取raft集群当前列表    
            //seata.server.raft.cluster=192.168.31.181:9091:8091:100,192.168.31.181:9092:8092:10,192.168.31.181:9093:8093:10
            String currentConf = ConfigurationFactory.getInstance().getConfig(SERVER_RAFT_CLUSTER);
            if (!StringUtils.isBlank(currentConf)) {
                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(group), currentClusters)) {
                	// 刷新路由表为最新的节点
                    routeTable.updateConfiguration(group, currentClusters);
                }
                try {
                    // 刷新路由表中的leader
                    routeTable.refreshLeader(RaftServerFactory.getCliClientServiceInstance(), group, 1000);
                    // 查询leader节点
                    PeerId leader = routeTable.selectLeader(group);
                    if (leader != null) {
                        Set<Node> nodes = new HashSet<>();
                        Node leaderNode = new Node();
                        // 添加leader
                        leaderNode.setRole(ClusterRole.LEADER);
                        leaderNode.setGroup(group);
                        leaderNode.setHttpPort(leader.getIdx() - SERVICE_OFFSET_SPRING_BOOT);
                        leaderNode.setNettyPort(leader.getIdx());
                        leaderNode.setHostAddress(leader.getIp());
                        nodes.add(leaderNode);
                        Configuration configuration = routeTable.getConfiguration(group);
                        // 添加学习者节点
                        nodes.addAll(configuration.getLearners().parallelStream().map(learner -> {
                            Node node = new Node();
                            node.setGroup(group);
                            node.setRole(ClusterRole.LEARNER);
                            node.setHttpPort(serverProperties.getPort());
                            node.setNettyPort(XID.getPort());
                            node.setHostAddress(learner.getIp());
                            return node;
                        }).collect(Collectors.toList()));
                        // 添加跟随者节点
                        nodes.addAll(configuration.getPeers().parallelStream().map(follower -> {
                            Node node = new Node();
                            node.setGroup(group);
                            node.setRole(ClusterRole.FOLLOWER);
                            node.setHttpPort(follower.getIdx() - SERVICE_OFFSET_SPRING_BOOT);
                            node.setNettyPort(follower.getIdx());
                            node.setHostAddress(follower.getIp());
                            return node;
                        }).collect(Collectors.toList()));
                        metadataResponse.setNodes(new ArrayList<>(nodes));
                    }
                } catch (Exception e) {
                    LOGGER.error("there is an exception to getting the leader address: {}", e.getMessage(), e);
                }
            }
        }
        return metadataResponse;
    }

    @GetMapping("/watch")
    public void watch(HttpServletRequest request, @RequestParam(defaultValue = DEFAULT_SEATA_GROUP) String groupIds,
        @RequestParam(defaultValue = "29000") int timeout, Long lastUpdateTime) {
        // 异步响应,获取异步上下文
        AsyncContext context = request.startAsync();
        // 不自动超时
        context.setTimeout(0L);
        for (String group : groupIds.split(",")) {
            // 创建观察者
            Watcher<AsyncContext> watcher = new Watcher<>(group, context, timeout, lastUpdateTime);
            // 注册观察者
            boolean success = clusterWatcherManager.registryWatcher(watcher);
            if (!success) {
                // 注册失败代表当前注册时已经发生了集群变化,直接推送让client刷新元数据
                HttpServletResponse httpServletResponse = (HttpServletResponse)watcher.getAsyncContext().getResponse();
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                context.complete();
            }
        }
    }

}

接下来我们看ClusterWatcherManager是如何管理观察者们的

@Component
public class ClusterWatcherManager implements ClusterChangeListener {
	// 每个group下的所有观察者
    private static final Map<String, Queue<Watcher<?>>> WATCHERS = new ConcurrentHashMap<>();
`	// 每个group 集群最新变化的时间
    private static final Map<String, Long> GROUP_UPDATE_TIME = new ConcurrentHashMap<>();
	// 定时处理超时需响应的请求
    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor =
        new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("long-polling", 1));

    @PostConstruct
    public void init() {
        // Responds to monitors that time out
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            for (String group : WATCHERS.keySet()) {
                // 直接remove整个队列,处理这批watcher时就不存在线程安全问题
                Optional.ofNullable(WATCHERS.remove(group))
                    .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
                        // 如果达到超时时间 那么将直接响应,让client重新watch
                        if (System.currentTimeMillis() > watcher.getTimeout()) {
                            HttpServletResponse httpServletResponse =
                                (HttpServletResponse)((AsyncContext)watcher.getAsyncContext()).getResponse();
                            watcher.setDone(true);
                            httpServletResponse.setStatus(HttpServletResponse.SC_NOT_MODIFIED);
                            ((AsyncContext)watcher.getAsyncContext()).complete();
                        }
                        // 如果watcher还未处理完成
                        if (!watcher.isDone()) {
                            // 那么重新注册
                            boolean success = registryWatcher(watcher);
                            if (!success) {
                                // 注册时恰好发生集群更新,那么直接响应httpstatus 200使client刷新metadata
                                HttpServletResponse httpServletResponse =
                                    (HttpServletResponse)((AsyncContext)watcher.getAsyncContext()).getResponse();
                                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                                ((AsyncContext)watcher.getAsyncContext()).complete();
                            }
                        }
                    }));
            }
        }, 1, 1, TimeUnit.SECONDS);
    }

    @Override
    @EventListener
    @Async
    public void onChangeEvent(ClusterChangeEvent event) {
        GROUP_UPDATE_TIME.put(event.getGroup(), System.currentTimeMillis());
        // Notifications are made of changes in cluster information
        // 直接remove整个队列,处理这批watcher时就不存在线程安全问题
        Optional.ofNullable(WATCHERS.remove(event.getGroup()))
            .ifPresent(watchers -> watchers.parallelStream().forEach(watcher -> {
                // 响应可刷新metadata予client
                HttpServletResponse httpServletResponse =
                    (HttpServletResponse)((AsyncContext)watcher.getAsyncContext()).getResponse();
                watcher.setDone(true);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                ((AsyncContext)watcher.getAsyncContext()).complete();
            }));
    }

    public boolean registryWatcher(Watcher<?> watcher) {
        String group = watcher.getGroup();
        Long lastChangeTime = GROUP_UPDATE_TIME.get(group);
        // watcher最后的最后时间大于等于集群变动的时间,那么说明client的数据已经是最新的,重新加回队列即可
        if (lastChangeTime == null || watcher.getLastUpdateTime() >= lastChangeTime) {
            WATCHERS.computeIfAbsent(group, value -> new ConcurrentLinkedQueue<>()).add(watcher);
        } else {
            return false;
        }
        return true;
    }

}

ok,client和server侧的元数据获取算是告一段落了,而看到这里大家会发现ClusterWatcherManager是一个ClusterChangeListener,那么到底是由哪里进行通知呢?

不急,我们先看看raft模式的初始化流程

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

// 预留给multi-raft使用,一个group,一个sessionmanager隔离
private static final Map<String, SessionManager> SESSION_MANAGER_MAP = new HashMap<>();

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)) {
            	// 如果是raft存储模式,那么将初始化raft专用的sessionmanager
               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;
           // 保存当前的raft group对应的sessionmanager
           SESSION_MANAGER_MAP.put(DEFAULT_SEATA_GROUP, 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 sessionMode the mode of store
    */
   protected static void reload(SessionMode sessionMode) {
       if (sessionMode == SessionMode.FILE) {
           ((Reloadable)ROOT_SESSION_MANAGER).reload();
           reload(ROOT_SESSION_MANAGER.allSessions(), sessionMode);
       } else {
           reload(null, sessionMode);
       }
   }

   public static void reload(Collection<GlobalSession> allSessions, SessionMode storeMode) {
       reload(allSessions, storeMode, true);
   }

   public static void reload(Collection<GlobalSession> allSessions, SessionMode storeMode, boolean acquireLock) {
       // raft节点选举完成后,leader会调此方法,但与file不同的是,file是启动时同步加载,需要竞争全局锁,而leader是通过raftlog恢复数据,自动会在
       // 恢复数据时竞争锁,故此处只在leader重新选举后调用,用于提前回滚还在active状态的事务,提前释放全局锁
       if ((SessionMode.FILE == storeMode || SessionMode.RAFT == storeMode)
           && CollectionUtils.isNotEmpty(allSessions)) {
           for (GlobalSession globalSession : allSessions) {
               GlobalStatus globalStatus = globalSession.getStatus();
               switch (globalStatus) {
                   case UnKnown:
                   case Committed:
                   case CommitFailed:
                   case Rollbacked:
                   case RollbackFailed:
                   case TimeoutRollbacked:
                   case TimeoutRollbackFailed:
                   case Finished:
                       removeInErrorState(globalSession);
                       break;
                   case AsyncCommitting:
                       queueToAsyncCommitting(globalSession);
                       break;
                   case Committing:
                   case CommitRetrying:
                       queueToRetryCommit(globalSession);
                       break;
                   default: {
                       if (acquireLock) {
                           lockBranchSessions(globalSession.getSortedBranches());
                           if (GlobalStatus.Rollbacking.equals(globalSession.getStatus())
                               || GlobalStatus.TimeoutRollbacking.equals(globalSession.getStatus())) {
                               globalSession.getBranchSessions().parallelStream()
                                   .forEach(branchSession -> branchSession.setLockStatus(LockStatus.Rollbacking));
                           }
                       }
                       switch (globalStatus) {
                           case Rollbacking:
                           case RollbackRetrying:
                           case TimeoutRollbacking:
                           case TimeoutRollbackRetrying:
                               queueToRetryRollback(globalSession);
                               break;
                           case Begin:
                               if (storeMode == SessionMode.RAFT) {
                                   // 活动中的事务肯定是还未全部解锁,将其只为回滚状态
                                   if (globalSession.isActive()) {
                                       globalSession.setStatus(GlobalStatus.RollbackRetrying);
                                       LOGGER.info("xid :{} change status RollbackRetrying", globalSession.getXid());
                                       queueToRetryRollback(globalSession);
                                   } else {
                                       // This means that the transaction is unlocked, but a reelection has occurred before the transaction state has been updated
                                       // so the transaction needs to be resubmitted (deleted)
                                       // 事务已经close了,说明已经释放了全局锁,可直接改为committing状态异步重试线程会删除这个事务
                                       globalSession.setStatus(GlobalStatus.Committing);
                                   }
                               } else {
                                   globalSession.setActive(true);
                               }
                               break;
                           default:
                               LOGGER.error("Could not handle the global session, xid: {}", globalSession.getXid());
                               throw new ShouldNeverHappenException("NOT properly handled " + globalStatus);
                       }
                       break;
                   }
               }
           }
       } else {
           // Redis, db and so on
           CompletableFuture.runAsync(() -> {
               SessionCondition searchCondition = new SessionCondition(GlobalStatus.UnKnown, GlobalStatus.Committed,
                       GlobalStatus.Rollbacked, GlobalStatus.TimeoutRollbacked, GlobalStatus.Finished);
               searchCondition.setLazyLoadBranch(true);

               long now = System.currentTimeMillis();
               List<GlobalSession> errorStatusGlobalSessions = ROOT_SESSION_MANAGER.findGlobalSessions(searchCondition);
               while (!CollectionUtils.isEmpty(errorStatusGlobalSessions)) {
                   for (GlobalSession errorStatusGlobalSession : errorStatusGlobalSessions) {
                       if (errorStatusGlobalSession.getBeginTime() >= now) {
                           // Exit when the global transaction begin after the instance started
                           return;
                       }

                       removeInErrorState(errorStatusGlobalSession);
                   }

                   // Load the next part
                   errorStatusGlobalSessions = ROOT_SESSION_MANAGER.findGlobalSessions(searchCondition);
               }
           });
       }
   }

   private static void removeInErrorState(GlobalSession globalSession) {
       try {
           LOGGER.warn("The global session should NOT be {}, remove it. xid = {}", globalSession.getStatus(), globalSession.getXid());
           // 失败异常的事务进行end处理,在raft下会删除,其余模式下将保留,待后续商讨
           getRootSessionManager().onSuccessEnd(globalSession);
           if (LOGGER.isInfoEnabled()) {
               LOGGER.info("Remove global session succeed, xid = {}, status = {}", globalSession.getXid(), globalSession.getStatus());
           }
       } catch (Exception e) {
           LOGGER.error("Remove global session failed, xid = {}, status = {}", globalSession.getXid(), globalSession.getStatus(), e);
       }
   }

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

public class RaftServerFactory {

    private static final Logger LOGGER = LoggerFactory.getLogger(RaftServerFactory.class);
    // multi-raft对应的raftserver
    private static final Map<String/*group*/, RaftServer/*raft-group-cluster*/> RAFT_SERVER_MAP = new HashMap<>();

    private Boolean raftMode = false;

    private static final io.seata.config.Configuration CONFIG = ConfigurationFactory.getInstance();

    public static RaftServerFactory getInstance() {
        return SingletonHandler.INSTANCE;
    }

    public static CliService getCliServiceInstance() {
        return SingletonHandler.CLI_SERVICE;
    }

    public static CliClientService getCliClientServiceInstance() {
        return SingletonHandler.CLI_CLIENT_SERVICE;
    }

    public void init() {
        // 获取集群初始化信息,也就是所有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);
        }
        int port = Integer.parseInt(System.getProperty(SERVER_RAFT_PORT_CAMEL, "0"));
        String host = XID.getIpAddress();
        int nettyPort = XID.getPort();
        PeerId serverId = null;
        if (port <= 0) {
            // 高可用时,不同的机器部署时,不需要设置 raft port,仅需在seata.server.raft.cluster中配置该节点信息会自动匹配
            // Highly available deployments require different nodes
            for (PeerId peer : initConf.getPeers()) {
                if (StringUtils.equals(peer.getIp(), host) && peer.getIdx() == nettyPort) {
                    serverId = peer;
                    break;
                }
            }
        } else {
            // 本地调试时使用,同个ip,不同的raft节点靠port区分
            // Local debugging use
            serverId = new PeerId(host, port, nettyPort);
        }
        // 获取当前的事务存储模式
        String mode = CONFIG.getConfig(ConfigurationKeys.STORE_MODE);
        StoreMode storeMode = StoreMode.get(mode);
        if (storeMode.equals(StoreMode.RAFT)) {
            raftMode = true;
        }
        // raft log&snapshot存储路径
        final String dataPath = CONFIG.getConfig(ConfigurationKeys.STORE_FILE_DIR, DEFAULT_SESSION_STORE_FILE_DIR)
            + separator + serverId.getPort();
        final NodeOptions nodeOptions = initNodeOptions(initConf);
        try {
            // as the foundation for multi raft group in the future
            RaftServer raftServer = new RaftServer(dataPath, DEFAULT_SEATA_GROUP, serverId, nodeOptions);
            // 预留给multi-raft使用
            RAFT_SERVER_MAP.put(DEFAULT_SEATA_GROUP, raftServer);
            LOGGER.info("started counter server at port:{}", raftServer.getNode().getNodeId().getPeerId().getPort());
        } catch (IOException e) {
            throw new IllegalArgumentException("fail init raft cluster:" + e.getMessage());
        }
        // whether to join an existing cluster
        // 如果开启自动加入集群,那么扩容时,无需改动seata.server.raft.cluster中的连接串,新节点会通过这个连接串将自己加入集群
        if (CONFIG.getBoolean(SERVER_RAFT_AUTO_JOIN, false)) {
            List<PeerId> currentPeers = null;
            try {
                currentPeers = getCliServiceInstance().getPeers(DEFAULT_SEATA_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(DEFAULT_SEATA_GROUP, initConf, serverId);
                    if (!status.isOk()) {
                        LOGGER.error("failed to join the RAFT cluster: {}. Please check the status of the cluster",
                            initConfStr);
                    }
                }
            }
        }
    }

    public RaftServer getRaftServer() {
        return getRaftServer(DEFAULT_SEATA_GROUP);
    }

    public RaftServer getRaftServer(String group) {
        return RAFT_SERVER_MAP.get(group);
    }

    public RaftStateMachine getStateMachine() {
        return getStateMachine(DEFAULT_SEATA_GROUP);
    }

    public RaftStateMachine getStateMachine(String group) {
        return RAFT_SERVER_MAP.get(group).getRaftStateMachine();
    }


    public Boolean isLeader() {
        return isLeader(DEFAULT_SEATA_GROUP);
    }

    public Boolean isLeader(String group) {
        // 获取对应的状态机,获取是否是leader
        AtomicReference<RaftStateMachine> stateMachine = new AtomicReference<>();
        Optional.ofNullable(RAFT_SERVER_MAP.get(group)).ifPresent(raftServer -> {
            stateMachine.set(raftServer.getRaftStateMachine());
        });
        RaftStateMachine raftStateMachine = stateMachine.get();
        return !isRaftMode() && RAFT_SERVER_MAP.isEmpty() || (raftStateMachine != null && raftStateMachine.isLeader());
    }

    public Boolean isRaftMode() {
        return raftMode;
    }

    public Boolean isNotRaftModeLeader() {
        return isNotRaftModeLeader(DEFAULT_SEATA_GROUP);
    }

    public Boolean isNotRaftModeLeader(String group) {
        return !isLeader(group) && isRaftMode();
    }

    private RaftOptions initRaftOptions() {
        RaftOptions raftOptions = new RaftOptions();
        raftOptions.setApplyBatch(CONFIG.getInt(SERVER_RAFT_APPLY_BATCH, raftOptions.getApplyBatch()));
        raftOptions.setMaxAppendBufferSize(
            CONFIG.getInt(SERVER_RAFT_MAX_APPEND_BUFFER_SIZE, raftOptions.getMaxAppendBufferSize()));
        raftOptions.setDisruptorBufferSize(
            CONFIG.getInt(SERVER_RAFT_DISRUPTOR_BUFFER_SIZE, raftOptions.getDisruptorBufferSize()));
        raftOptions.setMaxReplicatorInflightMsgs(
            CONFIG.getInt(SERVER_RAFT_MAX_REPLICATOR_INFLIGHT_MSGS, raftOptions.getMaxReplicatorInflightMsgs()));
        return raftOptions;
    }

    private NodeOptions initNodeOptions(Configuration initConf) {
        NodeOptions nodeOptions = new NodeOptions();
        // enable the CLI service.
        nodeOptions.setDisableCli(false);
        // snapshot should be made every 30 seconds
        Integer snapshotInterval = CONFIG.getInt(SERVER_RAFT_SNAPSHOT_INTERVAL, 60 * 10);
        nodeOptions.setSnapshotIntervalSecs(snapshotInterval);
        nodeOptions.setRaftOptions(initRaftOptions());
        // set the election timeout to 2 second
        nodeOptions
            .setElectionTimeoutMs(CONFIG.getInt(SERVER_RAFT_ELECTION_TIMEOUT_MS, DEFAULT_SERVER_RAFT_ELECTION_TIMEOUT_MS));
        // set up the initial cluster configuration
        nodeOptions.setInitialConf(initConf);
        return nodeOptions;
    }

    private static class SingletonHandler {
        // factory 单例
        private static final RaftServerFactory INSTANCE = new RaftServerFactory();
        // sofajraft cliservice 用作扩缩容使用
        private static final CliService CLI_SERVICE = RaftServiceFactory.createAndInitCliService(new CliOptions());
        // cli client seriver 理论上作用与client与server的路由寻址等,但seata设计为了减轻client负担,集成在了server侧
        private static final CliClientService CLI_CLIENT_SERVICE = new CliClientServiceImpl();
        static {
            CLI_CLIENT_SERVICE.init(new CliOptions());
        }
    }

}

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

public class RaftServer implements ConfigurationChangeListener, Disposable, Closeable {

    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 {
        // 该group的数据存储路径
        String groupPath = dataPath + File.separator + groupId;
        // Initialization path
        FileUtils.forceMkdir(new File(groupPath));
		// 选举通信使用的rpcserver
        // 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());
        // 初始化raft状态机
        // Initialize the state machine
        this.raftStateMachine = new RaftStateMachine(groupId);
        // Set the state machine to startup parameters
        nodeOptions.setFsm(this.raftStateMachine);
        // Set the storage path
        // Log, must
        nodeOptions.setLogUri(groupPath + File.separator + "log");
        // Meta information, must
        nodeOptions.setRaftMetaUri(groupPath + File.separator + "raft_meta");
        // Snapshot, optional, is generally recommended
        nodeOptions.setSnapshotUri(groupPath + File.separator + "snapshot");
        // raft metrics 开关
        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);
        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, 60),
                TimeUnit.MINUTES);
        }
    }

    public Node getNode() {
        return this.node;
    }

    public CliService getCliService() {
        return cliService;
    }

    public RaftStateMachine getRaftStateMachine() {
        return raftStateMachine;
    }

    @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();
                if (node != null && node.isLeader()) {
                    // 刷新集群列表
                    CliService cliService = getCliService();
                    cliService.changePeers(DEFAULT_SEATA_GROUP, getNode().getOptions().getInitialConf(), newConf);
                }
            }
        }
    }

    @Override
    public void close() {
        destroy();
    }

    @Override
    public void destroy() {
        if (raftGroupService != null) {
            raftGroupService.shutdown();
        }
        if (cliService != null) {
            cliService.shutdown();
        }
    }

}

ok,直此TC端的raft初始化相关内容已经解析完毕,核心的状态机将在下述存储模式篇章解析

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

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

RaftLockManager-Raft的全局锁实现,再老方案中一味的只考虑到follower因与leader执行相同的动作得到相同的结果,故忘记了动作可能存在无意义性,故在新版设计中,将raft的lockmanager直接继承file lock manager即可,并只需要在解锁时进行同步即可,在seata-server中file模式的锁是存在在内存中,也在branch身上,如果branch添加成功,那么锁一定存在,竞争锁的过程就不必同步了,竞争失败,branch一定不会添加成功,这是原子的.

@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(this.localReleaseGlobalSessionLock(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);
    }

    public boolean localReleaseGlobalSessionLock(GlobalSession globalSession) throws TransactionException {
        // raft的follower节点调用
        return super.releaseGlobalSessionLock(globalSession);
    }

    public boolean localReleaseLock(BranchSession branchSession) throws TransactionException {
        // raft的follower节点调用
        return super.releaseLock(branchSession);
    }

}

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

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

@LoadLevel(name = "raft", scope = Scope.PROTOTYPE)
public class RaftSessionManager extends FileSessionManager {

    public RaftSessionManager(String name) throws IOException {
        super(name);
    }

    @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());
        globalTransactionDO.setStatus(globalStatus.getCode());
        RaftSessionSyncMsg raftSyncMsg =
            new RaftSessionSyncMsg(UPDATE_GLOBAL_SESSION_STATUS, globalTransactionDO);
        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());
        branchTransactionDO.setStatus(branchStatus.getCode());
        RaftSessionSyncMsg raftSyncMsg =
                new RaftSessionSyncMsg(UPDATE_BRANCH_SESSION_STATUS, branchTransactionDO);
        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;
    }

}

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());
            }
        }
    }

}

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官网

public class RaftStateMachine extends StateMachineAdapter {

    private static final Logger LOGGER = LoggerFactory.getLogger(RaftStateMachine.class);
	// 当前事务模式
    private final String mode;
    // 当前raft group
    private final String group;

    private static final String BRANCH_SESSION_MAP_KEY = "branchSessionMap";

    private static final String GLOBAL_SESSION_MAP_KEY = "globalSessionMap";
	// raft同步消息的执行器
    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(String group) {
        this.group = group;
        mode = ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.STORE_MODE);
        // 当存储模式为raft,添加相关的执行器
        if (StoreMode.RAFT.getName().equalsIgnoreCase(mode)) {
            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
                // leader节点无需反序列化,而是本地就有该同步任务,直接运行即可
                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
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("sync msg: {}", msg);
                    }
                    // 执行同步消息的处理
                    onExecuteRaft(msg);
                }
            }
            iterator.next();
        }
    }

    @Override
    public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
        // 进入这里时状态机会阻塞,所以数据一定是一致的,线程安全
        if (!StringUtils.equals(StoreConfig.SessionMode.RAFT.getName(), mode)) {
            done.run(Status.OK());
            return;
        }
        // gets a record of the session at the moment
        Map<String, Object> maps = new HashMap<>(2);
        RaftSessionManager raftSessionManager = (RaftSessionManager)SessionHolder.getRootSessionManager();
        Map<String, GlobalSession> sessionMap = raftSessionManager.getSessionMap();
        int 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(GLOBAL_SESSION_MAP_KEY, globalSessionByteMap);
        maps.put(BRANCH_SESSION_MAP_KEY, branchSessionByteMap);
        // 创建raft快照
        RaftSnapshot raftSnapshot = new RaftSnapshot();
        // 将当前内存状态的事务信息放入body
        raftSnapshot.setBody(maps);
        LOGGER.info("globalSessionMap size :{}, branchSessionMap map size: {}", globalSessionByteMap.size(),
            branchSessionByteMap.size());
        // 获取snapshot存储路径
        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) {
        if (!StringUtils.equals(StoreConfig.SessionMode.RAFT.getName(), mode)) {
            return true;
        }
        // leader不需要加载快照,也不存在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(GLOBAL_SESSION_MAP_KEY);
            Map<Long, byte[]> branchSessionByteMap = (Map<Long, byte[]>)maps.get(BRANCH_SESSION_MAP_KEY);
            Map<String, GlobalSession> rootSessionMap = raftSessionManager.getSessionMap();
            // be sure to clear the data before loading it, because this is a full overwrite update
            // 由于快照可能是因为follower与leader拉开了较大差距,所以leader直接发送了一份最新数据的快照,所以要对内存已有数据清空
            LockerManagerFactory.getLockManager().cleanAllLocks();
            rootSessionMap.clear();
            // 快照内容是否为空
            if (!globalSessionByteMap.isEmpty()) {
                Map<String, GlobalSession> sessionMap = new HashMap<>();
                // 加载快照中的globalsession
                globalSessionByteMap.forEach((k, v) -> {
                    GlobalSession session = new GlobalSession();
                    session.decode(v);
                    sessionMap.put(k, session);
                });
                if (CollectionUtils.isNotEmpty(branchSessionByteMap)) {
                    // 加载快照中的branchsession
                    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())) {
                            globalSession.getBranchSessions().parallelStream()
                                    .forEach(branchSession -> branchSession.setLockStatus(LockStatus.Rollbacking));
                        }
                    });
                }
                // 将session装载到当前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) {
        boolean leader = isLeader();
        this.leaderTerm.set(term);
        super.onLeaderStart(term);
        // 允许事务协调者通过请求
        DefaultCoordinator.getInstance().setPrevent(true);
        // become the leader again,reloading global session
        if (!leader && RaftServerFactory.getInstance().isRaftMode()) {
            CompletableFuture.runAsync(() -> {
                LOGGER.info("session map: {} ", SessionHolder.getRootSessionManager().allSessions().size());
                // 这里由于发生了重选举,大概率还处于begin的事务决议会失败,将提前将其回滚,避免其占用的锁阻塞业务
                SessionHolder.reload(SessionHolder.getRootSessionManager().allSessions(), StoreConfig.SessionMode.RAFT,
                    false);
            });
        }
        // 这就是之前说的ClusterChangeListener(ClusterWatcherManager)的onChangeEvent从何而来,当节点成为leader时发出通知
        // watcher们就会收到节点变动而刷新client的metadata,后续与最新的leader进行通信
        ((ApplicationEventPublisher)ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
            .publishEvent(new ClusterChangeEvent(this, group));
    }

    @Override
    public void onLeaderStop(final Status status) {
        this.leaderTerm.set(-1);
        super.onLeaderStop(status);
        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);
        // 成为follower时代表要么是刚启动时,要么是出现了重选举,自己没选上或从leader退化为follower
        ((ApplicationEventPublisher)ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
            .publishEvent(new ClusterChangeEvent(this, group));
    }

    @Override
    public void onConfigurationCommitted(Configuration conf) {
        super.onConfigurationCommitted(conf);
        // 集群列表发生变化时,如添加一个节点成员或者删除一个,或者将follower变更为learner或者反之
        ((ApplicationEventPublisher)ObjectHolder.INSTANCE.getObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT))
            .publishEvent(new ClusterChangeEvent(this, group));
    }

    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

@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时,只有获取了锁成功的分支才会走到此处,所以当分支被同步的时候,也就意味着锁一定是争抢成功,这样就避免了很多无意义的锁同步行为,将锁同步和分支事务添加合并为一次同步任务即可.

@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,所以我们按顺序看响应的执行器

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;
    }
}

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;
    }

}

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;
    }
}

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状态机执行,将事务删除丢入异步线程执行,由于线程安全所以也不存在锁冲突
        RaftSessionManager raftSessionManager = (RaftSessionManager) SessionHolder.getRootSessionManager(sessionSyncMsg.getGroup());
        GlobalSession globalSession =
                raftSessionManager.findGlobalSession(sessionSyncMsg.getGlobalSession().getXid());
        try {
            raftLockManager.localReleaseGlobalSessionLock(globalSession);
        } catch (TransactionException e) {
            logger.error(e.getMessage(), e);
        }
        EXECUTOR.execute(() -> {
            GlobalSession globalSession =
                raftSessionManager.findGlobalSession(sessionSyncMsg.getGlobalSession().getXid());
            if (globalSession != null) {
                try {
                    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;
    }

}

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;
    }

}

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());
            }
            // 为了解决先解锁后leader突然宕机,导致无法区分事务是要回滚还是提交的问题,只要全部释放了锁,就无所谓提交还是回滚事务的状态了,可直接提交
            globalSession.setActive(false);
            // 释放全局锁
            return raftLockManager.localReleaseGlobalSessionLock(globalSession);
        }
        return false;
    }

}

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

3.2.4.Client&Server端通信解析

如上图所示,client请求至server,再从serveronrequestprocessor进而找到对应的处理器,最终通过raftsessionmanager存储事务信息,而事务信息通过raft协议,从各个seataserver的状态机中执行事务的存储动作,最终再响应至client.

看过了watch和querymetadata的同学相信会有一个疑问,就是如果client侧获取最新metadata时,恰好有业务线程在begin或者commit或者registry的动作会发生什么?它们所请求的leader很可能在这一刻已经不存在或者不是leader了,那么前者比较简单rpc就会失败,后者在tc侧做了一个当前是否是leader的检测,如果不是leader将拒绝该请求,如果是leader但在中途失败,那么会由于当前已经不是leader,createTask提交到状态机的动作会失败,那么client侧也会接收到响应异常,而该旧leader的提交任务也会失败,也避免了事务信息不一致的问题

@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);
    }
}

public void setPrevent(boolean prevent) {
    // 只有raft模式才需要变动prevent
    if (StoreConfig.getSessionMode() == StoreConfig.SessionMode.RAFT) {
        this.prevent = prevent;
    }
}

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

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

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

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

/**
 * 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下定时任务是如何处理的.

/**
 * 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处理的

@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模式

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

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

答: 1.性能考虑: 分布式事务的二阶段下,明显带来的就是多次rpc确认事务决议状态,并下发响应结果造成的性能下降,如果直接由server端进行转发,那么其实也存在转发过程中leader的变更,这种由server端转发的方案会带来一定的性能下降,所以目前没有此考虑. 2.中间件职责不同: zookeeper这类的工具并不会像seata那样会有大量的读写请求(其实有,但是性能非常不理想,这也是为什么那么多开源中间件开始放弃zk的原因之一),即便使用的tps低一些也是能接受.再者在2022年年底宣布KRaft集群模式生产可用(GA)的Kafka而言,其也是通过metadata方式刷新client本地的路由,client的消息也是从metadata中找到分区leader进行发送,如果分区leader重选举会刷新元数据重试,及rocketmq也有类似的功能,定时从namingserver中拉取元数据.

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模式,且也能保证数据的一致性.

6.在raft这类事务结束时就将事务信息删除的模式,如何追溯历史数据?

答:

一. 社区目前已有完成的将事务全部流转状态(增删改)全部发送至kafka中的pr,业务侧可消费此topic将事务存储至hive或Cassandra等数据中间件中持久化存储,可做报表或查询平台功能.

二. 社区未来可能针对此问题引入rocksdb或h2这类本地存储中,将事务的删除转为删除+存储至rocksdb或h2中方便事务的回溯查询

5.总结

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

并针对业界发展趋势,如clickhouse和kafka都开始抛弃zk,转为自研的如clickkeeper和KRaft,将元数据等信息交由自身保证存储,减少第三方依赖减少运维成本和学习成本,这也是非常成熟和可借鉴效仿的特性.

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

赏

谢谢你请我吃糖果

支付宝

微信