跳到主要内容

· 阅读需 11 分钟

seata目前支持多种注册中心的实现,为了提供整个链路的闭环功能,seata设计推出了原生的注册中心namingserver

2. 领域模型

2.1 命名空间与事务分组

  • Namespace:在NamingServer模型中,命名空间(Namespace)用于实现命名空间的环境隔离。它允许在不同的环境(如开发、测试、生产)中隔离各自的服务实例。
  • Cluster与Unit:Cluster(集群)负责事务分组的处理,Unit则在每个集群内部做负载均衡。事务分组(vgroup)在命名空间和集群的配合下,通过元数据定位到具体的TC节点。

2.2 事务处理流程与namingserver的交互

img 事务处理流程与namingserver的交互的流程如下:

1.在client侧配置好Namingserver的地址和相关配置

2.client启动后TM向namingserver发起服务发现的请求

3.namingserver根据TM传来的vGroup参数和内存中的事务分组映射关系返回相关的集群列表,namingserver返回的集群列表元数据如下

{
"clusterList": [
{
"clusterName": "cluster2",
"clusterType": "default",
"groupList":[group1,group2]
"unitData": [
{
"unitName": "115482ee-cf27-45d6-b17e-31b9e2d7892f",
"namingInstanceList": [
{
"ip": "172.31.31.191",
"port": 8092,
"nettyPort": 0,
"grpcPort": 0,
"weight": 1.0,
"healthy": true,
"timeStamp": 1695042063334,
"role": member,
"metadata": {
"weight": 1,
"cluster-type": "default"
}
}
]
},
{
"unitName": "097e6ab7-d2d2-47e4-a578-fae1a4f4c517",
"namingInstanceList": [
{
"ip": "172.31.31.191",
"port": 8091,
"nettyPort": 0,
"grpcPort": 0,
"weight": 1.0,
"healthy": true,
"timeStamp": 1695042076481,
"role": member,
"metadata": {
"weight": 1,
"cluster-type": "default"
}
}
]
}
]
}
],
"term": 1695042076578
}

4.客户端通过负载均衡策略找出合适的TC节点开启事务

5.TM将事务分组和TC节点传递给RM

6.RM向TC节点发起分支注册的请求

7.TC节点完成二阶段下发

3.设计思路

3.1 AP还是CP?

CAP协议又称CAP定理,指的是在一个分布式系统中,Consistency(一致性)、 Availability(可用性)、Partition tolerance(分区容错性),三者不可得兼。 分布式系统的CAP理论:理论首先把分布式系统中的三个特性进行了如下归纳:

● 一致性(C):在分布式系统中的所有数据备份,在同一时刻是否同样的值。(等同于所有节点访问同一份最新的数据副本)

● 可用性(A):在集群中一部分节点故障后,集群整体是否还能响应客户端的读写请求。(对数据更新具备高可用性)

● 分区容错性(P):以实际效果而言,分区相当于对通信的时限要求。系统如果不能在时限内达成数据一致性,就意味着发生了分区的情况,必须就当前操作在C和A之间做出选择。

对于namingserver而言,我们更倾向于使用AP模型,即注重可用性和分区容错性,牺牲一定的一致性。NamingServer作为服务注册中心,主要职责是提供高效的服务发现与注册服务,而对短时间内的数据一致性要求可以适当放松。在分布式环境中,可能会出现多个节点短暂的注册数据不一致现象。例如,当多个NamingServer节点发生了网络分区时,某些节点获取的注册信息可能有延迟。 对于NamingServer来说,这种短暂的不一致性我们认为是可以容忍的。由于服务注册与发现的强一致性要求并不高,即使在某个时刻有部分节点的注册数据滞后或不一致,也不会立刻影响到整个系统的正常服务。通过心跳机制、周期性同步等方式,最终一致性可以逐渐得到保证。

3.2 Quorum NWR机制在NamingServer中的应用

Quorum NWR(仲裁读写)是一种在分布式系统中用于确保数据一致性的机制。该机制通过设置副本的总数(N)、写操作需要成功的副本数(W)、读操作需要访问的副本数(R)来协调数据的一致性。在NamingServer的设计中,采用了多写+补偿机制来保证多个NamingServer节点之间的信息一致性,而客户端则与某一个NamingServer节点交互以获取注册信息。

  1. 写入操作(W-写仲裁): 当有集群节点变化时,server端会将请求发送到NamingServer集群中的多个节点。 根据NWR机制,系统确保至少有W个副本成功写入注册信息。 通过多写机制,即使某些节点暂时不可用或存在网络延迟,仍能确保写操作的高可用性。一旦W个节点写入成功,客户端将收到成功响应。 补偿机制:对于没有立即成功写入的副本,系统会通过异步补偿方式,在稍后的时间段内同步这些节点,确保最终一致性。
  2. 读取操作(R-读仲裁): 客户端通过与NamingServer集群中的任意一个节点进行交互,发送读取请求以获取服务注册信息。 系统会从至少R个副本中读取数据,采用最新版本的数据作为返回结果。即使某些节点的数据存在短暂不一致,客户端通过读取多个副本并比较其版本号或一致性标记,能够确保读取到最新的注册信息。 由于客户端只与一个NamingServer节点进行交互,读取操作的效率得到了提升,并且避免了多个节点之间的复杂协调,系统依然能保证最终一致性。
  3. NWR参数的设计与权衡: 在namingserver中我们设置W=N、R=1。W=N虽然意味着写入需要发送到所有节点,但并不要求所有节点必须立即成功写入。系统允许某些节点暂时失败,通过补偿机制在后续阶段同步这些节点,从而提高系统的容错性,因为即便某些节点在写入时发生故障或网络中断,数据更新仍然可以通过补偿机制最终传播到所有节点。这既保证了系统的高可用性,也确保了数据最终在所有节点上是一致的。由于写操作要求所有节点参与,因此每个节点都会接收到最新的数据更新。 客户端在执行读取操作时,可以从任意一个NamingServer节点读取数据,而不必担心数据不一致的问题。即使某些节点未能在写入时立即成功,客户端仍能从其他已成功写入的节点获取最新的注册信息。这样,R值可以设定较低(如R=1),从而提高读取操作的效率,同时系统通过补偿机制确保所有节点最终达到一致。 img

3.2 架构图

img

namingserver的运行链路如上图所示:

  1. 通过控制台在某个cluster下创建一个事务分组vgroup。
  2. 创建vgroup->cluster的请求发送给namingserver,namingserver再传递给对应的tc节点。
  3. tc节点将vgroup->cluster映射关系持久化保存。
  4. tc节点在心跳的时候将vgroup->cluster的映射关系更新到所有的namingserver。
  5. client通过自己配置的事务分组vgroup从namingserver获取对应cluster元数据。
  6. client在事务流程中,使用cluster下的unit进行负载均衡,再进行begin,registry,commit,rollback等。
  7. 事务决议后,对应的unit下的leader节点下发二阶段,无状态节点下,每个unit的唯一node就是leader。

3.3 设计细节

3.3.1 长轮询推送集群变化情况

img

如上图所示,每隔30s client侧需要向namingserver发起一次服务发现的请求,用以拉取最新的tc列表。而在这30s的间隔中,client侧将采用HTTP长轮询的方式一直watch namingserver节点,如果namingserver 侧有如下的变化:

1.事务分组映射关系的变化;

2.集群中实例的增加或者减少;

3.集群中实例的属性的变化;

那么watch返回200状态码,告知client需要获取最新集群信息;否则namingserver将一直挂起watch方法,直到HTTP长轮询超时,然后返回304状态码, 告知client进行下一轮watch。

· 阅读需 14 分钟

目前seata支持丰富的第三方配置中心,但是考虑使用的便捷性同时为了降低使用seata的门槛,在seata-server利用现有的sofa-jraft+rocksdb构建一个配置中心功能,seata-client直接与seata-server通信,获取seata相关的配置,不需要再去第三方配置中心读取,实现配置中心自闭环。

2. 设计说明

2.1 配置中心

在现有的第三方配置中心实现中,Client端和Server端对于配置中心是解耦的,Client端和Server端直接通过Configuration实例获取配置项,且Configuration对于Client端和Server端的初始化行为是一致的,都是先连接到配置中心中间件然后获取配置,以及添加监听器等。 img

当使用Raft的实现配置中心后,由于所有的配置项信息是保存在Server端的,因此初始化Configuration实例时对于Client端和Server端的行为是不一致的。

此外为保证和原来配置中心的逻辑相同,Client端和Server端获取配置项依旧统一从RaftConfiguration实例中获取,不直接和RocksDB进行交互。 img

img

RaftConfiguration分为Server端和Client端,按照启动环境返回不同配置实例。

public class RaftConfigurationProvider implements ConfigurationProvider {
@Override
public Configuration provide() {
String applicationType = System.getProperty(APPLICATION_TYPE_KEY);
if (APPLICATION_TYPE_SERVER.equals(applicationType)){
return RaftConfigurationServer.getInstance();
}else{
return RaftConfigurationClient.getInstance();
}
}
}

@SpringBootApplication(scanBasePackages = {"org.apache.seata"})
public class ServerApplication {
public static void main(String[] args) throws IOException {
System.setProperty(APPLICATION_TYPE_KEY, APPLICATION_TYPE_SERVER);
// run the spring-boot application
SpringApplication.run(ServerApplication.class, args);
}
}

2.2 配置存储模块

img

抽象设计

为了未来支持和拓展更多的KV内存键值对数据库(如LevelDB,Caffeine),现抽象一个ConfigStoreManager接口以及抽象类AbstractConfigStoreManager,提供如下公共方法:

  • Get:获取指定namespace,dataId中名为key的单一配置项
  • GetAll:获取指定namespace,dataId中全部配置项
  • Put:新增/修改指定namespace,dataId中某一配置项<key,value>
  • Delete:删除指定namespace,dataId中名为key的配置项
  • DeleteAll:删除指定namespace,dataId中全部配置项
  • Clear:清空所有配置
  • GetAllNamespaces:获取所有命名空间
  • GetAllDataIds:获取指定namespace下的所有配置dataIds
  • ...

ConfigStoreManagerFactory、ConfigStoreManagerProvider.java:使用SPI机制实现的配置存储工厂类和提供者。

配置监听

Server端和Client端配置中心需要监听配置项的变化。

在Server端,由于配置本身存储在Server端,我们直接拦截配置变更的方法即可。我们在抽象接口中定义了addConfigListenerremoveConfigListener方法用户添加和删除配置监听器。监听的逻辑由具体的实现类负责。

RocksDBConfigStoreManager中,定义了notifyConfigChange方法来触发监听,当调用写相关操作时(如Put、Delete)触发该方法来通知配置的变更。从而触发回调事件通知Server端配置中心。

在Client端,我们定义了配置版本号长连接机制来实现配置的监听。具体的Client端在启动时,与Server建立长连接并定期刷新该连接。Server端内部维护一个watchMap存放所有客户端的监听信息。每当Raft状态机执行配置更新的操作会发送一个ApplicationEvent事件,该事件被ClusterConfigWatcherManager监听,从而通知watchMap中所有客户端配置变更。此外使用了配置版本号来优化实现,在建立长连接时,客户端需要传入版本号,当版本号低于Server端对应配置的版本号时直接返回最新配置。反之,若Server端配置版本号低于本地,则Client端认为该Server的配置已经过期(可能宕机或集群发生脑裂)会重试请求集群中其他的节点。

多租户方案

在Seata-Server端存放配置时,需要实现多租户的配置隔离,要求不同租户间的配置是相互独立、物理/逻辑上是隔离的。

首先调研了业内使用RocksDB的开源项目的实现,总结如下。

  1. JRaft,单RocksDB实例,两个列族,一个用来存Raft条目,一个用来存元信息。
  2. TiKV,两个RocksDB实例,分别为raftDB,kvDB。kvDB中使用了多个列族存放元数据,用户数据,锁数据等。
  3. Pika,为每个数据结构(String,Hash,List,Set,Zset等)创建了一个RocksDB实例,每个RocksDB实例分别用多个列族存储数据,比如Data、Meta

考虑到无法提前知道租户数量(无法在启动时创建指定数量的RocksDB),因此使用单个RocksDB实例,多列族存储。不同租户使用namespace区分,在Rocksdb中通过列族(ColumnFamily)进行逻辑隔离,一个namespace对应一个列族。列族相当于关系型数据库中表的概念。在配置的增删改查时指定namespace操作具体的列族,实现多租户的隔离。此外名为config_version的列族是内置的,用于对现有的配置进行版本号跟踪。 img

3. 使用方式

3.0 准备配置文件

首先准备好配置文件,具体可以参考:配置文件示例。 并将配置文件置于Seata server项目资源目录下。

3.1 Server端配置

application.yml 中加入Raft配置中心配置,其余配置参考

config:
# support: nacos, consul, apollo, zk, etcd3, raft
type: raft
raft:
db:
type: rocksdb # db类型,目前只支持rocksdb
dir: configStore # db文件存储目录
destroy-on-shutdown: false #应用关闭时是否清除db文件, 默认false
namespace: 'default' # 命名空间
dataId: 'seata.properties' # 配置文件id
file:
name: 'file' # 初始配置文件名

server:
raft:
group: default #此值代表该raft集群的group,client的事务分组对应的值要与之对应
server-addr: 192.168.241.1:9091, 192.168.241.2:9091 ,192.168.241.3:9091 # 其他Raft节点的ip和端口,端口为该节点的netty端口+1000,默认netty端口为8091
snapshot-interval: 600 # 600秒做一次数据的快照,以便raftlog的快速滚动,但是每次做快照如果内存中事务数据过多会导致每600秒产生一次业务rt的抖动,但是对于故障恢复比较友好,重启节点较快,可以调整为30分钟,1小时都行,具体按业务来,可以自行压测看看是否有抖动,在rt抖动和故障恢复中自行找个平衡点
apply-batch: 32 # 最多批量32次动作做一次提交raftlog
max-append-bufferSize: 262144 #日志存储缓冲区最大大小,默认256K
max-replicator-inflight-msgs: 256 #在启用 pipeline 请求情况下,最大 in-flight 请求数,默认256
disruptor-buffer-size: 16384 #内部 disruptor buffer 大小,如果是写入吞吐量较高场景,需要适当调高该值,默认 16384
election-timeout-ms: 1000 #超过多久没有leader的心跳开始重选举
reporter-enabled: false # raft自身的监控是否开启
reporter-initial-delay: 60 # 监控的区间间隔
serialization: jackson # 序列化方式,不要改动
compressor: none # raftlog的压缩方式,如gzip,zstd等
sync: true # raft日志的刷盘方式,默认是同步刷盘

在Seata-Server下,需要一个初始配置文件作为Server端的配置文件(也就是上一步所述的配置文件),file.name配置项需要和该文件名保持一致。在Server初次启动时,会将该配置文件作为Raft配置中心的初始配置。目前支持的文件类型有:conf、yaml、properties、txt。

注意:Raft集群内节点的初始配置文件需要保持一致。

3.2 控制台配置管理界面

当Server端使用Raft模式的配置中心后,可通过在Seata Console中内置的配置管理页面,进行配置中心的管理。用户可以通过在该页面对存储于Seata-Server集群的配置进行增删改查,注意该操作是对于集群生效的,因此可以在集群中的任意节点进行修改,所有操作会通过Raft在集群内进行同步。

注意:该配置管理页面仅在配置中心为Raft模式下开启,对其他类型的配置中心不开放。

3.2.1 配置隔离

Raft配置中心提供了namespace命名空间的机制来实现多租户的配置隔离。不同namespace下的配置通过底层存储机制实现逻辑隔离。在同一namespace下可以有多套配置文件,不同的配置文件用dataId进行区分。一套配置以namespacedataId来唯一标识。

例如:

  • namespace=default(默认),dataId=seata.properties(默认)

  • namespace=dev,dataId=seata-server.properties,dataId=seata-client.yaml

  • namespace=prop,dataId=seata-server.properties,dataId=seata-client.txt

img

3.2.2 配置上传

在Sever启动时,Server端配置的初始文件会自动上传到配置中心。此外用户还可以通过点击"上传(Upload)"按钮上传配置文件到指定的namespacedataId。 当配置上传到Server端配置中心后,Client端就可以通过namespacedataId来获取具体的配置文件了。

img

目前支持上传的配置文件类型有:txt、text、yaml、properties类型。具体的配置文件可参考示例:配置文件示例

3.2.3 配置查询

选择namespacedataId后,可点击"搜索(Search)"按钮查询该配置下的所有配置项信息。

配置以配置项列表的形式呈现,每一行都代表一个配置项,以KeyValue展示。

img

3.2.4 配置删除

当不再需要某一套配置时,用户可以删除指定namespacedataId的配置数据。

注意该操作一旦完成,会清空该配置下的所有配置项信息,且无法恢复。请避免删除正在使用中的配置。

img

3.2.5 配置修改

在配置项列表中,用户可以对该配置下的某个配置项进行新增修改删除操作。操作成功后Server端和Client端会及时收到配置变更,从而获取到最新的值。

  • 新增:在当前配置下添加配置项

img

  • 修改:修改指定配置项的值。

img

  • 删除:删除指定的配置项

img

3.3 Client端配置

Client需要添加如下的配置项。其中raft.server-addr需要和Server端Raft集群的IP地址列表一致。

config:
type: raft # Raft模式
raft:
server-addr: 192.168.241.1:7091, 192.168.241.2:7091 ,192.168.241.3:7091 # 配置raft相关元数据的获取地址
username: 'seata' # 鉴权
password: 'seata' # 鉴权
db:
namespace: 'default' # 命名空间
dataId: 'seata.properties' # 配置文件Id

此外,client需要引入HttpClient依赖,用于通过Http请求向Seata-Server集群获取配置信息

<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>

配置完成后,Client应用启动时就会从raft.server-addr配置的Server中订阅并获取指定namespacedataId的配置,并通过监听机制在配置发生变更时获取获取最新配置。

· 阅读需 9 分钟

引言和概述

在分布式系统中,通信协议的设计直接影响系统的可靠性和可扩展性。Apache Seata的RPC通信协议为各组件间的数据传输提供了基础,对这方面的源码分析是深入理解seata的又一个好方式。在最近的2.2.0版本里,我为Seata的通信机制进行了重构,以支持多版本协议的兼容性,现在改造完成了,我将从传输机制和通信协议两个方面去分析新版本里源码。 本文是第一篇,介绍Seata传输机制。

seata里的RPC通信主角是TCTMRM三者,当然过程中还可能会涉及注册中心甚至配置中心等其他网络交互,但这些相对内容所使用的通信机制是相对独立的,本文不作讨论。

接下来我将按照我最早了解源码时的几个直觉提问,带大家进行探索。

Seata中的Netty(谁在传输)

第一个问题:seata通信的底层是什么在进行发送请求报文和接收请求报文?答案是netty,而netty在seata里面是如何工作的呢?我们将去到core包的org.apache.seata.core.rpc.netty去探索

从这个继承关系我们可以看到,AbstractNettyRemoting作为核心的父类,RM和TM和Server(TC)都实现了他,实际上这个类里面已经实现了核心的发送和接收

sendSync实现了同步发送逻辑,异步发送sendAsync的逻辑相似且更简单这里不再重复,只要拿到channel进行发送即可

protected Object sendSync(Channel channel, RpcMessage rpcMessage, long timeoutMillis) throws TimeoutException {
// 此处省略非关键代码

MessageFuture messageFuture = new MessageFuture();
messageFuture.setRequestMessage(rpcMessage);
messageFuture.setTimeout(timeoutMillis);
futures.put(rpcMessage.getId(), messageFuture);

channelWritableCheck(channel, rpcMessage.getBody());

String remoteAddr = ChannelUtil.getAddressFromChannel(channel);
doBeforeRpcHooks(remoteAddr, rpcMessage);

// netty的写方法(netty write)
channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
MessageFuture messageFuture1 = futures.remove(rpcMessage.getId());
if (messageFuture1 != null) {
messageFuture1.setResultMessage(future.cause());
}
destroyChannel(future.channel());
}
});

try {
Object result = messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);
doAfterRpcHooks(remoteAddr, rpcMessage, result);
return result;
} catch (Exception exx) {
// 此处省略非关键代码
}
}

而接收报文的方式,主要在processMessage方法里,这个方法被AbstractNettyRemotingClient.ClientHandlerAbstractNettyRemotingServer.ServerHandler这两个类的channelRead调用,这两个内部类都是ChannelDuplexHandler的子类,他们各自注册在client和server的Bootstrap里(为什么注册到bootstrap就能进行接收操作?这个要移步netty的原理)

收到信息后就会调进父类的processMessage方法里,我们来看看源码

protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
// 此处省略非关键代码
Object body = rpcMessage.getBody();
if (body instanceof MessageTypeAware) {
MessageTypeAware messageTypeAware = (MessageTypeAware) body;
final Pair<RemotingProcessor, ExecutorService> pair = this.processorTable.get((int) messageTypeAware.getTypeCode());
if (pair != null) {
// 这里可读性稍微差点,first是Processor,表示普通处理,而second是线程池,表示池化处理。
if (pair.getSecond() != null) {
try {
pair.getSecond().execute(() -> {
try {
pair.getFirst().process(ctx, rpcMessage);
} catch (Throwable th) {
LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
} finally {
MDC.clear();
}
});
} catch (RejectedExecutionException e) {
// 此处省略非关键代码
}
} else {
try {
pair.getFirst().process(ctx, rpcMessage);
} catch (Throwable th) {
LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
}
}
} else {
LOGGER.error("This message type [{}] has no processor.", messageTypeAware.getTypeCode());
}
} else {
LOGGER.error("This rpcMessage body[{}] is not MessageTypeAware type.", body);
}
}

实际上这些processor和executor是client和server注册进来的处理器:下面是一部分的处理器,他们对应着不同的MessageType,以下是部分处理器的注册举例(他们在NettyRemotingServer#registerProcessor)

        super.registerProcessor(MessageType.TYPE_GLOBAL_ROLLBACK, onRequestProcessor, messageExecutor);
super.registerProcessor(MessageType.TYPE_GLOBAL_STATUS, onRequestProcessor, messageExecutor);
super.registerProcessor(MessageType.TYPE_SEATA_MERGE, onRequestProcessor, messageExecutor);
super.registerProcessor(MessageType.TYPE_BRANCH_COMMIT_RESULT, onResponseProcessor, branchResultMessageExecutor);
super.registerProcessor(MessageType.TYPE_BRANCH_ROLLBACK_RESULT, onResponseProcessor, branchResultMessageExecutor);
super.registerProcessor(MessageType.TYPE_REG_RM, regRmProcessor, messageExecutor);
super.registerProcessor(MessageType.TYPE_REG_CLT, regTmProcessor, null);

可以看到这些processor实际上就是seata各种提交回滚等等的处理器

Seata中的NettyChannel(channel怎么管理)

那第二个问题,既然上面是netty依靠着channel在进行着收发,那这个channel怎么来呢?会一直持有吗?如果断了怎么重连?答案在ChannelManager和上面的两个reg的processor

当RM/TM取得了server的地址,进行注册的时候(第一次通信),如果server能成功解析报文并发现是REG信息,就会进入regRmProcessor/regTmProcessor,这里以TM为例子

// server接收 RegTmProcessor
private void onRegTmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
RegisterTMRequest message = (RegisterTMRequest) rpcMessage.getBody();
String ipAndPort = NetUtil.toStringAddress(ctx.channel().remoteAddress());
Version.putChannelVersion(ctx.channel(), message.getVersion());
boolean isSuccess = false;
String errorInfo = StringUtils.EMPTY;
try {
if (null == checkAuthHandler || checkAuthHandler.regTransactionManagerCheckAuth(message)) {
// 在ChannelManager中注册channel,这里可以预见到注册之后,server进行sendSync(channel,xxx)的时候就可以拿到这个channel了
ChannelManager.registerTMChannel(message, ctx.channel());
Version.putChannelVersion(ctx.channel(), message.getVersion());
isSuccess = true;
}
} catch (Exception exx) {
isSuccess = false;
errorInfo = exx.getMessage();
LOGGER.error("TM register fail, error message:{}", errorInfo);
}
RegisterTMResponse response = new RegisterTMResponse(isSuccess);
// 异步回复
remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
// 此处省略非关键代码
}

// ChannelManager
public static void registerTMChannel(RegisterTMRequest request, Channel channel)
throws IncompatibleVersionException {
RpcContext rpcContext = buildChannelHolder(NettyPoolKey.TransactionRole.TMROLE, request.getVersion(),
request.getApplicationId(),
request.getTransactionServiceGroup(),
null, channel);
rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
String clientIdentified = rpcContext.getApplicationId() + Constants.CLIENT_ID_SPLIT_CHAR + ChannelUtil.getClientIpFromChannel(channel);
ConcurrentMap<Integer, RpcContext> clientIdentifiedMap = CollectionUtils.computeIfAbsent(TM_CHANNELS, clientIdentified, key -> new ConcurrentHashMap<>());
rpcContext.holdInClientChannels(clientIdentifiedMap);
}

在ChannelManager里管理着RM_CHANNELSRM_CHANNELS两个比较复杂的map,特别是RM_CHANNELS里面有4层(resourceId -> applicationId -> ip -> port -> RpcContext)

说完了server对channel的管理,那client呢?这个map管理更简单一些,就是注册成功后在onRegisterMsgSuccess里面也用一个NettyClientChannelManager里registerChannel,后续跟server交互尽量都用这个channel。

那么第三个问题又来了,client的channel不可用了可以自行新建,可是server接收后发现这是新channel怎么办?或者server在异步回复的时候发现channel不可用了怎么办? 答案依然在NettyClientChannelManager,这里面相对复杂的是,client方面需要用到channel的时候,实际上由一个对象池nettyClientKeyPool管理着,这是个apache的objectPool,所以当channel不可用时也会由这个池子去新增并在使用完后入池。这个对象池实际上是一直持有着RegisterTMRequest,跟第一次进来时一样,每次创建需要创建channel的时候,实际上都发生了一次注册

// NettyClientChannelManager
public Channel makeObject(NettyPoolKey key) {
InetSocketAddress address = NetUtil.toInetSocketAddress(key.getAddress());
if (LOGGER.isInfoEnabled()) {
LOGGER.info("NettyPool create channel to " + key);
}
Channel tmpChannel = clientBootstrap.getNewChannel(address);
Object response;
Channel channelToServer = null;
// key持有RegisterTMRequest
if (key.getMessage() == null) {
throw new FrameworkException("register msg is null, role:" + key.getTransactionRole().name());
}
try {
// 实际上是在进行reg操作
response = rpcRemotingClient.sendSyncRequest(tmpChannel, key.getMessage());
if (!isRegisterSuccess(response, key.getTransactionRole())) {
rpcRemotingClient.onRegisterMsgFail(key.getAddress(), tmpChannel, response, key.getMessage());
} else {
channelToServer = tmpChannel;
// 成功后会入池
rpcRemotingClient.onRegisterMsgSuccess(key.getAddress(), tmpChannel, response, key.getMessage());
}
}
// 此处省略非关键代码

return channelToServer;
}

最后

这一篇我们了解了seata是怎样借助netty来传输数据的,为了更好的看懂netty处理的全貌,我画了个层级图

上面已经讲了请求发送时,serverHandler/clientHandler和NettyRemoting(包括RM、TM、TC)的处理,知道了从外部到netty处理器再到内部的DefaultCoordinator的过程,但我们还缺Decoder/Encoder没讲,这里面会进行协议的解析/封装,也会进行序列化和反序列化,请看 Seata的RPC通信源码分析02:协议篇

· 阅读需 11 分钟

引言和概述

上一篇Seata的RPC通信源码分析01:传输篇已经介绍了RPC通信的传输机制,这一篇我们继续来看协议部分的内容,把这个图里没解析清楚的encode/decode部分给补充完整。

同样的,我们以提问来深入的方式去探究它。在本文中,我们不仅要了解二进制如何解析成rpcMsg类型,还要知道如何兼容不同版本的协议,那么第一个问题:协议长什么样?

协议结构

上图展示了协议在0.7.1之前和之后的变化,(在ProtocolDecoderV1的注释也可以看到,更旧版本的要看ProtocolV1Decoder),可以看到新版本的有以下这些构成部分

  • magic-code:0xdada
  • protocal-verson:版本号
  • full-length:总长度
  • head-length:头部长度
  • msgtype:消息类型
  • serializer/codecType:序列化方式
  • compress:压缩方式
  • requestid:请求id

这里我们说明一下seata各版本的server之间对协议的处理差异

  • version<0.7.1 : 只能处理v0版本的协议(上图中的上半部分,带有flag段的),无法识别其他版本协议
  • 0.7.1<=version<2.2.0 : 只能处理v1版本的协议(上图中的下半部分),无法识别其他版本协议
  • version>=2.2.0 : 可以同时识别v0和v1版本的协议,并处理

那么2.2.0是怎样做到兼容的呢?先卖个关子,在说明这个之前我们先看看v1的encoder和decoder分别都是怎样运作的。需要注意的是,和之前提到的传输机制一样,协议处理也是client和server共用的,所以下面提到的都是通用逻辑。

从ByteBuf到RpcMessage(Encoder/Decoder做了什么)

先来看ProtocolDecoderV1

    public RpcMessage decodeFrame(ByteBuf frame) {
byte b0 = frame.readByte();
byte b1 = frame.readByte();

// 获取version
byte version = frame.readByte();
// 获取header和body以外的字段
int fullLength = frame.readInt();
short headLength = frame.readShort();
byte messageType = frame.readByte();
byte codecType = frame.readByte();
byte compressorType = frame.readByte();
int requestId = frame.readInt();

ProtocolRpcMessageV1 rpcMessage = new ProtocolRpcMessageV1();
rpcMessage.setCodec(codecType);
rpcMessage.setId(requestId);
rpcMessage.setCompressor(compressorType);
rpcMessage.setMessageType(messageType);

// 头部信息
int headMapLength = headLength - ProtocolConstants.V1_HEAD_LENGTH;
if (headMapLength > 0) {
Map<String, String> map = HeadMapSerializer.getInstance().decode(frame, headMapLength);
rpcMessage.getHeadMap().putAll(map);
}

// 如果是心跳信息不需要对body进行序列化
if (messageType == ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST) {
rpcMessage.setBody(HeartbeatMessage.PING);
} else if (messageType == ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE) {
rpcMessage.setBody(HeartbeatMessage.PONG);
} else {
int bodyLength = fullLength - headLength;
if (bodyLength > 0) {
byte[] bs = new byte[bodyLength];
frame.readBytes(bs);
// 根据刚才得到的compressorType来按需做解压处理
Compressor compressor = CompressorFactory.getCompressor(compressorType);
bs = compressor.decompress(bs);
SerializerType protocolType = SerializerType.getByCode(rpcMessage.getCodec());
if (this.supportDeSerializerTypes.contains(protocolType)) {
// 序列化器,由于这个是v1专用的ProtocolDecoderV1,所以可以直接传入version1
Serializer serializer = SerializerServiceLoader.load(protocolType, ProtocolConstants.VERSION_1);
rpcMessage.setBody(serializer.deserialize(bs));
} else {
throw new IllegalArgumentException("SerializerType not match");
}
}
}
return rpcMessage.protocolMsg2RpcMsg();
}

由于encode操作正好和decode操作相反,这里不再重复介绍,我们继续看里面的serialize操作。上面的serialize类来自SerializerServiceLoader

    public static Serializer load(SerializerType type, byte version) throws EnhancedServiceNotFoundException {
// 先处理PROTOBUF方式的序列化,通过反射工具取得
if (type == SerializerType.PROTOBUF) {
try {
ReflectionUtil.getClassByName(PROTOBUF_SERIALIZER_CLASS_NAME);
} catch (ClassNotFoundException e) {
throw new EnhancedServiceNotFoundException("'ProtobufSerializer' not found. " +
"Please manually reference 'org.apache.seata:seata-serializer-protobuf' dependency ", e);
}
}

String key = serialzerKey(type, version);
// 这里是一个SERIALIZER_MAP,相当于序列化类的缓存。为什么需要缓存,因为SeataSerializer的scope = Scope.PROTOTYPE,防止多次创建类
Serializer serializer = SERIALIZER_MAP.get(key);
if (serializer == null) {
if (type == SerializerType.SEATA) {
// 这里是seata的SPI机制,本文不再往里深入加载类的逻辑,只需要知道去加载Serializer这个接口,并且把version给到了构造方法
serializer = EnhancedServiceLoader.load(Serializer.class, type.name(), new Object[]{version});
} else {
serializer = EnhancedServiceLoader.load(Serializer.class, type.name());
}
SERIALIZER_MAP.put(key, serializer);
}
return serializer;
}

// 这里是SeataSerializer构造方法,里面是单例模式的构造器,因为现在是两个版本各一个类,也可以说是双例
public SeataSerializer(Byte version) {
if (version == ProtocolConstants.VERSION_0) {
versionSeataSerializer = SeataSerializerV0.getInstance();
} else if (version == ProtocolConstants.VERSION_1) {
versionSeataSerializer = SeataSerializerV1.getInstance();
}
if (versionSeataSerializer == null) {
throw new UnsupportedOperationException("version is not supported");
}
}

这样,decoder就得到了一个Serializer,程序运行到rpcMessage.setBody(serializer.deserialize(bs)),我们来看看deserialize是怎样处理的

    public <T> T deserialize(byte[] bytes) {
return deserializeByVersion(bytes, ProtocolConstants.VERSION_0);
}
private static <T> T deserializeByVersion(byte[] bytes, byte version) {
//前面是合法性判断,此处忽略
ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
short typecode = byteBuffer.getShort();
ByteBuffer in = byteBuffer.slice();
//创建父类,并根据版本号创建Codec
AbstractMessage abstractMessage = MessageCodecFactory.getMessage(typecode);
MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typecode, version);
//codec的decode操作
messageCodec.decode(abstractMessage, in);
return (T) abstractMessage;
}

很遗憾,这个serialize并没有太多逻辑,关键还是在MessageCodecFactory和Codec,我们继续往里看。可以看到MessageCodecFactory内容不少,但形式单一,都是根据MessageType返回message和codec,所以这里不再展示factory的内容,我们直接看message和codec,也就是messageCodec.decode(abstractMessage, in),虽然codec类型还是很多,但我们可以看到他们的结构都是相似的,逐个字段解析:

    // BranchRegisterRequestCodec的decode,这个请求是注册一个事务分支
public <T> void decode(T t, ByteBuffer in) {
BranchRegisterRequest branchRegisterRequest = (BranchRegisterRequest)t;

// 解析xid
short xidLen = in.getShort();
if (xidLen > 0) {
byte[] bs = new byte[xidLen];
in.get(bs);
branchRegisterRequest.setXid(new String(bs, UTF8));
}
// 解析branchType
branchRegisterRequest.setBranchType(BranchType.get(in.get()));
short len = in.getShort();
if (len > 0) {
byte[] bs = new byte[len];
in.get(bs);
branchRegisterRequest.setResourceId(new String(bs, UTF8));
}
// 解析lockKey
int iLen = in.getInt();
if (iLen > 0) {
byte[] bs = new byte[iLen];
in.get(bs);
branchRegisterRequest.setLockKey(new String(bs, UTF8));
}
// 解析applicationData
int applicationDataLen = in.getInt();
if (applicationDataLen > 0) {
byte[] bs = new byte[applicationDataLen];
in.get(bs);
branchRegisterRequest.setApplicationData(new String(bs, UTF8));
}
}

好了,到这里,我们已经得到了branchRegisterRequest,可以愉快地交给TCInboundHandler处理了。

但是问题又来了,我们只看到client(RM/TM)有以下这种添加encoder/decoder的代码,也就是我们知道client都使用当前版本的encoder/decoder处理:

        bootstrap.handler(
new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new IdleStateHandler(nettyClientConfig.getChannelMaxReadIdleSeconds(),nettyClientConfig.getChannelMaxWriteIdleSeconds(),nettyClientConfig.getChannelMaxAllIdleSeconds()))
.addLast(new ProtocolDecoderV1())
.addLast(new ProtocolEncoderV1());
if (channelHandlers != null) {
addChannelPipelineLast(ch, channelHandlers);
}
}
});

但server如何处理?还有说好的多版本协议呢?

多版本协议(版本识别和绑定)

我们先来看encoder/decoder的一个类图:

ProtocolDecoderV1我们已经分析完了,ProtocolEncoderV1是反向操作,应该比较好理解,至于ProtocolDecoderV0和ProtocolEncoderV0,从图上也可以看到他们和v1是平行关系,除了v0的操作(虽然目前为止我们还没让他派上用场),他们都是netty里典型的encode和decode的子类,但MultiProtocolDecoder又是什么?他是多版本协议的主角,而且在启动的时候已经注册进server的bootstrap。

    protected boolean isV0(ByteBuf in) {
boolean isV0 = false;
in.markReaderIndex();
byte b0 = in.readByte();
byte b1 = in.readByte();
// 实际上,识别协议就靠第3个byte(b2),只要是正常的新版本,b2就是大于0的版本号,而对于0.7以下的版本来说,b2是FLAG的第一位,正好无论是哪种情况他都是0
// v1/v2/v3 : b2 = version
// v0 : b2 = 0 ,1st byte in FLAG(2byte:0x10/0x20/0x40/0x80)
byte b2 = in.readByte();
if (ProtocolConstants.MAGIC_CODE_BYTES[0] == b0 && ProtocolConstants.MAGIC_CODE_BYTES[1] == b1 && 0 == b2) {
isV0 = true;
}
// 读完的字节还要吐回去,为了让各版本的decoder能从头解析
in.resetReaderIndex();
return isV0;
}
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
ByteBuf frame;
Object decoded;
byte version;
try {
// 识别版本号,获取当前版本号
if (isV0(in)) {
decoded = in;
version = ProtocolConstants.VERSION_0;
} else {
decoded = super.decode(ctx, in);
version = decideVersion(decoded);
}

if (decoded instanceof ByteBuf) {
frame = (ByteBuf) decoded;
ProtocolDecoder decoder = protocolDecoderMap.get(version);
ProtocolEncoder encoder = protocolEncoderMap.get(version);
try {
if (decoder == null || encoder == null) {
throw new UnsupportedOperationException("Unsupported version: " + version);
}
// 首次进来,使用判断好的decoder进行操作
return decoder.decodeFrame(frame);
} finally {
if (version != ProtocolConstants.VERSION_0) {
frame.release();
}
// 首次进来,绑定对应version的encoder和decoder,也就相当于绑定了channel
ctx.pipeline().addLast((ChannelHandler)decoder);
ctx.pipeline().addLast((ChannelHandler)encoder);
if (channelHandlers != null) {
ctx.pipeline().addLast(channelHandlers);
}
// 绑定好之后,将自身移除,后续不再判断
ctx.pipeline().remove(this);
}
}
} catch (Exception exx) {
LOGGER.error("Decode frame error, cause: {}", exx.getMessage());
throw new DecodeException(exx);
}
return decoded;
}

protected byte decideVersion(Object in) {
if (in instanceof ByteBuf) {
ByteBuf frame = (ByteBuf) in;
frame.markReaderIndex();
byte b0 = frame.readByte();
byte b1 = frame.readByte();
// 和isV0方法相似,取第3个byte
if (ProtocolConstants.MAGIC_CODE_BYTES[0] != b0 || ProtocolConstants.MAGIC_CODE_BYTES[1] != b1) {
throw new IllegalArgumentException("Unknown magic code: " + b0 + ", " + b1);
}

byte version = frame.readByte();
frame.resetReaderIndex();
return version;
}
return -1;
}

通过上面的分析,v0终于派上用场(当有旧版本的client注册时,server就会为其分配低版本的encoder/decoder),我们也摸清了多版本协议如何识别、如何绑定。

· 阅读需 10 分钟

背景

随着 Seata 项目的不断发展和壮大,我们的贡献者群体也在持续扩大。项目的功能不断增强,对于代码质量的要求也在提高。在这个过程中,我们期望每一位贡献者在提交功能代码的同时,能够附带规范、完备的测试用例。

一个优秀的项目,其完备的单元测试是基本保障。Test-Driven Development(TDD)理念已经提出多年,它强调在编写功能代码之前先编写测试用例。通过编写单元测试,开发者可以更深入地理解代码中相关类和方法的作用,掌握核心逻辑,熟悉各种场景的运行情况。同时,单元测试也为开源项目提供了稳定安全的保障,使得项目在接受贡献者代码时,能够确保代码的质量和稳定性。 单元测试是质量保障的第一环,有效的单元测试能够提前发现90%以上的代码Bug问题,同时也能防止代码的腐化。在项目重构和演进过程中,单元测试起到了至关重要的作用,它能够确保重构后的代码仍然能够正常工作,不会引入新的Bug。

在社区看来,贡献合理的测试用例代码和贡献功能代码同样重要,为了帮助开发者编写出高质量的测试用例,本文给出一些基础的规范和建议。

推荐的框架

当前社区使用以下三个框架编写测试用例;

junit5

junit是Java中最常用的单元测试框架,用于编写和运行可重复的测试用例。

        <junit-jupiter.version>5.8.2</junit-jupiter.version>
<dependency>
<groupId>org.junit</groupId>
<artifactId>junit-bom</artifactId>
<version>${junit-jupiter.version}</version>
</dependency>

mockito

mockito是一个mock框架,主要是用来做mock测试,他可以模拟任何 Spring管理的 bean、模拟方法的返回值、模拟抛出异常等,可以让我们在缺乏一些依赖的情况下,完成测试及验证。

        <mockito.version>4.11.0</mockito.version>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>${mockito.version}</version>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-junit-jupiter</artifactId>
<version>${mockito.version}</version>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-inline</artifactId>
<version>${mockito.version}</version>
</dependency>

assertj

assertj是一个断言库,提供了一组易于使用和可读性很强的断言方法,当junit的断言难以满足时,可以使用assertj进行断言;

请注意:我们在seata-dependencies的pom.xml中统一管理了这三个库的版本。

        <assertj-core.version>3.12.2</assertj-core.version>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>${assertj-core.version}</version>
</dependency>

规范

我们参考阿里巴巴JAVA开发手册,整理了一些建议及规范,分为不同的级别,其中【【强制】部分,开发者需要严格遵守,社区在合并代码时会按照强制规则进行review,【【推荐】【参考】部分,方便大家更好的了解我们对于测试用例的考量和原则。

1.【强制】单元测试必须遵守 AIR 原则。

说明:好的单元测试宏观上来说,具有自动化、独立性、可重复执行的特点。

  • A:Automatic(自动化)
  • I:Independent(独立性)
  • R:Repeatable(可重复)
2.【强制】单元测试应该是全自动执行的,并且非交互式的。

测试用例通常是被定期执行的,执行过程必须完全自动化才有意义。输出结果需要人工检查的测试不是一个好的单元测试。单元测试中不准使用 System.out 来进行人肉验证,必须使用 assert 来验证。

3.【强制】保持单元测试的独立性。为了保证单元测试稳定可靠且便于维护,单元测试用例之间决不能互相调用,也不能依赖执行的先后次序。

反例:method2 需要依赖 method1 的执行,将执行结果作为 method2 的输入。

4.【强制】单元测试是可以重复执行的,不能受到外界环境的影响。

说明:单元测试通常会被放到持续集成中,每次有代码 check in 时单元测试都会被执行。如果单测对外部环境(网络、服务、中间件等)有依赖,容易导致持续集成机制的不可用。

正例:为了不受外界环境影响,要求设计代码时就把 SUT 的依赖改成注入,在测试时用 spring 这样的 DI框架注入一个本地(内存)实现或者 Mock 实现。

5.【强制】对于单元测试,要保证测试粒度足够小,有助于精确定位问题。单测粒度至多是类级别,一般是方法级别。

说明:只有测试粒度小才能在出错时尽快定位到出错位置。单测不负责检查跨类或者跨系统的交互逻辑,那是集成测试的领域。

6.【强制】核心业务、核心应用、核心模块的增量代码确保单元测试通过。

说明:新增代码及时补充单元测试,如果新增代码影响了原有单元测试,请及时修正。

7.【强制】单元测试代码必须写在如下工程目录:src/test/java,不允许写在业务代码目录下。

说明:源码编译时会跳过此目录,而单元测试框架默认是扫描此目录。

8.【强制】单元测试的基本目标:语句覆盖率达到 70%;核心模块的语句覆盖率和分支覆盖率都要达到 100%。

说明:在工程规约的应用分层中提到的 DAO 层,Manager 层,可重用度高的 Service,都应该进行单元测试。

9.【推荐】编写单元测试代码遵守 BCDE 原则,以保证被测试模块的交付质量。
  • B:Border,边界值测试,包括循环边界、特殊取值、特殊时间点、数据顺序等。
  • C:Correct,正确的输入,并得到预期的结果。
  • D:Design,与设计文档相结合,来编写单元测试。
  • E:Error,强制错误信息输入(如:非法数据、异常流程、业务允许外等),并得到预期的结果。
10.【推荐】对于数据库相关的查询,更新,删除等操作,不能假设数据库里的数据是存在的,或者直接操作数据库把数据插入进去,请使用程序插入或者导入数据的方式来准备数据。

反例:删除某一行数据的单元测试,在数据库中,先直接手动增加一行作为删除目标,但是这一行新增数据并不符合业务插入规则,导致测试结果异常。

11.【推荐】和数据库相关的单元测试,可以设定自动回滚机制,不给数据库造成脏数据。或者对单元测试产生的数据有明确的前后缀标识。
12.【推荐】对于不可测的代码在适当的时机做必要的重构,使代码变得可测,避免为了达到测试要求而书写不规范测试代码。
13.【推荐】单元测试作为一种质量保障手段,在提交pr前完成单元测试的编写及验证。
14.【参考】为了更方便地进行单元测试,业务代码应避免以下情况:
  • 构造方法中做的事情过多。
  • 存在过多的全局变量和静态方法。
  • 存在过多的外部依赖。
  • 存在过多的条件语句。 说明:多层条件语句建议使用卫语句、策略模式、状态模式等方式重构。

· 阅读需 20 分钟

Seata 是一款开源的分布式事务解决方案,致力于在微服务架构下提供高性能和简单易用的分布式事务服务。在今年的开源之夏活动中,我加入了 Apache Seata (Incubator) 社区,完成了开源之夏的课题,并从此一直积极参与社区。我有幸在云栖大会-开发者秀场上分享了我的开发者经验。在本文中,我将与大家分享我在 Seata 社区中的开发者之旅,以及在这个旅程中积累的经验和见解。希望通过我的故事,能够激励更多人踏上这充满挑战和激励的开源之路,为开源社区的繁荣做出自己的贡献。

相关背景

在正式介绍我的经历之前,我想先提供一些相关的背景信息,以解释为什么我要参与开源以及如何参与开源。关于参与开源的原因,我相信每个人都有不同的动机。以下是我认为一些主要的原因:

  • 学习:参与开源使我们有机会为不同组织开发的开源项目做出贡献,与行业专家互动,提供了学习的机会。
  • 技能提升:以我为例,我通常使用 Java 和 Python 进行后端开发。但在参与Seata项目时,我有机会学习Go语言,拓宽了我的后端技术栈。此外作为学生,我很难接触到生产级框架或应用,而开源社区为我提供了这个机会。
  • 兴趣:我身边的朋友都是热衷于开源的,他们享受编程,对开源充满热情。
  • 求职:参与开源可以丰富我们的作品集,为简历增加分量。
  • 工作需求:有时参与开源是为了解决工作中遇到的问题或满足工作需求。

这些都是参与开源的原因,对我来说,学习、技能提升和兴趣是我参与开源的主要动机。无论你是在校学生还是在职人员,如果你有参与开源的意愿,不要犹豫,任何人都可以为开源项目做出贡献。年龄、性别、工作和所在地都不重要,关键是你的热情和对开源项目的好奇心。

我参与开源的契机是参加了中科院软件所举办的开源之夏活动。

开源之夏是一个面向高校开发者的开源活动,社区发布开源项目,学生开发者在导师的指导下完成项目的开发,结项成果贡献给社区,合入社区仓库,获得项目奖金和证书。开源之夏是踏入开源社区的一个绝佳契机,也是我第一次比较正式地接触开源项目,而这个经历为我打开了一扇全新的大门。自此我深刻地认识到参与开源项目的建设,分享自己的技术成果,让更多的开发者能够使用你所贡献的东西,是一件极富乐趣和意义的事情。

下面我分享的这张图片是开源之夏官方公开的数据,从 2020 年开始参与的社区数量还有学生数量都在逐年增加,活动也是越办越好。可以看到今年的参与的社区项目共有 133 个,每个社区又提供了若干个课题,而每位学生只能选择一个课题。想要在这么多个社区中找到想要参与的社区和适合自己的课题是一个相对复杂的任务。

img

综合考虑社区的活跃程度、技术栈契合度、新人引导情况等,最终我选择加入 Seata 社区。

Seata 是一款开源的分布式事务框架,提供了完整的分布式事务解决方案,包括 AT、TCC、Saga 和 XA 事务模式,可支持多种编程语言和数据存储方案。从 19 年开源起到今年已经走过了 5 个年头,社区中有超过 300 多位贡献者,项目收获了 24k+ 星标,是一个非常成熟的社区。同时 Seata 兼容 10 余种主流 RPC 框架和 RDBMS,与 20 多个社区存在集成和被集成的关系,被几千家客户应用到业务系统中,可以说是分布式事务解决方案的事实标准。

img

2023 年 10 月 29 日,Seata 正式捐赠给了 Apache 软件基金会,成为孵化项目。经过孵化之后,Seata将有望成为首个 Apache 软件基金会的分布式事务框架顶级项目。这次捐赠也将推动 Seata 更广泛地发展,对生态系统的建设产生深远的影响,从而使更多的开发者受益。这个重要的里程碑也为 Seata 带来更广阔的发展空间。

开发之旅

介绍完了一些基本情况,后文中我将分享我在 Seata 社区的开发之旅。

在正式开始开发之前,我进行了许多准备工作。因为 Seata 已经经历了五年的发展,积累了数十万行代码,因此直接参与开发需要一定的上手成本。我分享了一些准备经验,希望能够为大家提供一些启发。

  1. 文档和博客是第一手材料 文档和博客这类的文本材料可以帮助社区新人迅速了解项目背景和代码结构。 首先,官方文档是最主要的参考资料,从这里可以了解到一切官方认为你需要了解的东西。 img 博客,仅次于官方文档的材料,一般是开发者或者是深度用户编写的,和文档不同的点在于博客可能会更深入到某个专项上去介绍,比如一些项目的理论模型、项目结构、某个模块的源码分析等等。 img 公众号,和博客类似,一般是偏技术性的文章,公众号还有个优点是可以订阅推送,利用碎片时间阅读一些技术。 img 此外,开源社区的一些在线分享或线下 Meetup 公开的幻灯片也是非常有意义的文本资料。 img 除了官方资料之外,还有许多第三方资料可供学习,比如可以通过用户分享的 use cases 了解项目的具体实施和实践;通过第三方社区的集成文档了解项目的生态;还有就是通过第三方的视频教程来学习。但在所有这些资料中,我认为官方文档和博客是最有帮助的。
  2. 熟悉使用框架 当然刚才说的这些文本资料肯定不需要面面俱到的看完,纸上得来终觉浅,看到感觉差不多明白了就可以去实践了。可以按照官方文档的"Get Started"章节逐步了解项目的基本流程。另一种方法是查找官方提供的示例或演示,构建并运行它们,理解代码和配置的含义,并通过使用项目了解项目的需求、目标以及现有功能和架构。 例如,Seata有一个名为 seata-samples 的仓库,其中包含20多种用例,比如 Seata 和 Dubbo 集成,和 SCA, Nacos 集成的案例,基本可以覆盖到支持的所有场景。
  3. 粗略阅读源代码把握主要逻辑 在准备阶段,粗略地阅读源代码以把握项目的主要逻辑也很重要。了解如何高效地把握项目的主要内容是一个需要长期积累的技能。首先,通过前述的准备步骤,了解项目的概念、交互和流程模型是很有帮助的。 以Seata为例,通过官方文档和实际操作,可以了解Seata事务领域的三个角色:TC(Transaction Coordinator)、TM(Transaction Manager)和 RM(Resource Manager)。TC 作为独立部署的 Server 用于维护全局和分支事务的状态,是 Seata 实现高可用的关键;TM 用于与 TC 交互,定义全局事务的开始、提交或回滚;RM 用于管理分支事务处理的资源,与 TC 交互以注册分支事务和报告分支事务的状态,并驱动分支事务提交或回滚。粗略地了解这些角色之间的交互后,可以更轻松地把握项目的主要逻辑。 img 脑海里刻下了这些模型的印象,对源码的主干提取就相对得心应手了一些。比如 Seata TC 事务协调者,作为 Server 端,是一个独立于业务部署的单独应用。那为了分析源码,就可以直接在本地把 server 起起来,通过启动类开始追踪。可以分析到一些初始化的逻辑比如服务注册、全局锁的初始化等等。还有可以通过 RPC 的调用来追踪到交互逻辑的代码,比如 TC 是如何对全局事务和分支事务进行持久化,如何驱动全局事务提交或者回滚的。 然而内嵌客户端的框架代码,没有一个启动类入口可以入手分析。那其实可以从一个 sample 入手,找到其对框架代码的引用从而进行阅读。比如 Seata 一个很重要的注解是 GlobalTransaction,用于标识一个全局事务。想要知道 TM 是如何对这个注解分析的,那我们通过 IDE 的搜索功能,找到 GlobalTransaction 的拦截器即可分析其中的逻辑。 还有一个小 tips 分享给大家,往往来说单测注重于单一模块的职能,可以通过阅读单测可以了解一个模块的输入输出、逻辑边界,也可以顺着单测的调用链去阅读代码,也是理解源码一个很重要的手段。

万事俱备只欠东风,做完充足的准备,下一步就是区积极参与到社区之中。

参与的方式也有很多种,最常见的参与方式是查看项目的 Issues 列表,社区通常会为新贡献者标记一些带有特殊标签的 Issue,如“good-first-issue”、“contributions-welcome”和“help-wanted”等。可以通过这些标签筛选感兴趣的任务。

img

除了 Issues,GitHub 还提供了讨论的功能,可以参与一些公开的讨论并获取新的想法。

img

此外,社区通常会定期举行会议,比如周会或双周会,可以通过参加这些会议来了解社区的最新进展,提出问题以及与其他社区成员交流。

总结与心得

我加入 Seata 社区最初是通过开源之夏活动。我完成了我的课题,为 Seata Saga 实现了一些新的功能,也做了一系列的优化。但我不止于此,因为在 Seata 的开源经历中我获得了学生生涯中最宝贵的一次开发者体验,在之后的时间我也持续通过上述参与方式持续活跃在社区中。这主要得益于以下几个方面:

  1. 沟通与社交:导师制度为我提供了重要的支持。在开发过程中,我与我的导师亦夏之间的密切合作对我适应社区文化和工作流程起到了关键作用。他不仅帮助我适应了社区,还为我提供了程序设计的思路,也与我分享了一些在工作中的经验和见解,这些都对我的发展非常有帮助。此外,Seata 社区创始人清铭也提供了很多帮助,包括建立了与其他同学的联系,帮助我进行 Code Review,也为我提供了许多机会。
  2. 正反馈:在 Seata 的开发过程中,我经历了一个良性的循环。许多细节为我提供了许多正反馈,例如我的贡献能被用户广泛使用和受益,比如开发得到了社区的认可。这些正反馈加强了我继续在 Seata 社区贡献的意愿。
  3. 技能提升:再就是参与 Seata 开发,对我能力的提升也是巨大的。在这里,我能学习到生产级别的代码,包括性能优化,接口设计,边界判断的技巧。可以直接参与一个开源项目的运作,包括项目计划,安排,沟通等。当然还了解一个分布式事务框架是如何设计并实现的。

除了这些宝贵的开发者体验,我也从这次经历中体悟到了一些关于参与开源的个人心得,为激励其他有兴趣参与开源社区的同学,我做了简单的总结:

  1. 了解和学习社区文化和价值观:每个开源社区都有不同的文化和价值观。了解社区的文化和价值观对于成功参与社区至关重要。观察和了解社区其他成员的日常开发和交流方式是学习社区文化的好方法。在社区中要尊重他人的意见和包容不同的观点。
  2. 敢于迈出第一步:不要害怕面对困难,迈出第一步是参与开源社区的关键。可以通过领取标有"good-first-issue"等标签的 Issue,编写文档、单元测试等方式来开始。重要的是要克服畏难情绪,积极尝试并学习。
  3. 对自己的工作要充满信心:不要怀疑自己的能力。每个人都是从零开始的,没有人天生就是专家。参与开源社区是一个学习和成长的过程,需要不断的实践和积累经验。
  4. 积极参与讨论,持续学习不同技术:不要害怕提出问题,无论是关于项目的具体技术还是开发过程中的挑战。同时也不要局限于一个领域。尝试学习和掌握不同编程语言、框架和工具,这可以拓宽技术视野,为项目提供有价值的洞见。

通过我的开源之旅,我积累了宝贵的经验和技能,这些不仅帮助我成长为一个更有价值的开发者,也让我深刻地了解了开源社区的力量。然而,我不仅仅是个别的参与者,我代表着 Seata 社区的一部分。Seata 作为一个正在不断成长和演变的开源项目,有着巨大的潜力,同时也面临着新的挑战。因此我要强调 Seata 社区的重要性和未来的潜力,它已经进入 Apache 软件基金会的孵化阶段,这个重要的里程碑将为 Seata 带来更广阔的发展空间。Seata 欢迎更多的开发者和贡献者的加入,让我们共同推动这个开源项目的发展,为分布式事务领域的进步贡献一份力量。

· 阅读需 24 分钟

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

目前Seata的分布式事务数据存储模式有file,db,redis,而本篇文章将Seata-Server Raft模式的架构,部署使用,压测对比,及为什么Seata需要Raft,并领略从调研对比,设计,到具体实现,再到知识沉淀的过程.

分享人:陈健斌(funkye) github id: funky-eyes

2. 架构介绍

2.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.2 为什么需要raft模式

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

2.2.1 现有存储模式

在当前的 Seata 设计中,Server 端的作用是保证事务的二阶段被正确执行。然而,这取决于事务记录的正确存储。为确保事务记录不丢失,需要在保持状态正确的前提下,驱动所有的 Seata-RM 执行正确的二阶段行为。那么,Seata 目前是如何存储事务状态和记录的呢?

首先介绍一下 Seata 支持的三种事务存储模式:file、db 和 redis。根据一致性的排名,db 模式下的事务记录可以得到最好的保证,其次是 file 模式的异步刷盘,最后是 redis 模式下的 aof 和 rdb

顾名思义:

  • file 模式是 Seata 自实现的事务存储方式,它以顺序写的形式将事务信息存储到本地磁盘上。为了兼顾性能,默认采用异步方式,并将事务信息存储在内存中,确保内存和磁盘上的数据一致性。当 Seata-Server(TC)意外宕机时,在重新启动时会从磁盘读取事务信息并恢复到内存中,以便继续运行事务上下文。
  • db 是 Seata 的抽象事务存储管理器(AbstractTransactionStoreManager)的另一种实现方式。它依赖于数据库,如 PostgreSQL、MySQL、Oracle 等,在数据库中进行事务信息的增删改查操作。一致性由数据库的本地事务保证,数据也由数据库负责持久化到磁盘。
  • redis 和 db 类似,也是一种事务存储方式。它利用 Jedis 和 Lua 脚本来进行事务的增删改查操作,部分操作(如竞争锁)在 Seata 2.x 版本中全部采用了 Lua 脚本。数据的存储与 db 类似,依赖于存储方(Redis)来保证数据的一致性。与 db 类似,redis 在 Seata 中采用了计算和存储分离的架构设计.

2.2.2 高可用

高可用简单理解就是集群能够在主节点宕机后继续正常运行,常见的方式是通过部署多个提供相同服务的节点,并通过注册中心实时感知主节点的上下线情况,以便及时切换到可用的节点。

看起来似乎只需要加几台机器进行部署,但实际上背后存在一个问题,即如何确保多个节点像一个整体一样运作。如果其中一个节点宕机,另一个节点能够完美接替宕机节点的工作,包括处理宕机节点的数据。解决这个问题的答案其实很简单,在计算与存储分离的架构下,只需将数据存储在共享的存储中间件中,任何一个节点都可以通过访问该公共存储区域获取所有节点操作的事务信息,从而实现高可用的能力。

然而,前提条件是计算与存储必须分离。为什么计算与存储一体化设计不可行呢?这就要说到 File 模式的实现了。如之前描述的,File 模式将数据存储在本地磁盘和节点内存中,数据写操作没有任何同步,这意味着目前的 File 模式无法实现高可用,仅支持单机部署。作为初级的快速入门和简单使用而言,File 模式适用性较低,高性能的基于内存的 File 模式也基本上不再被生产环境使用。

2.3 Seata-Raft是如何设计的呢?

2.3.1 设计原理

Seata-Raft模式的设计思路是通过封装无法高可用的file模式,利用Raft算法实现多个TC之间数据的同步。该模式保证了使用file模式时多个TC的数据一致性,同时将异步刷盘操作改为使用Raft日志和快照进行数据恢复。 流程图

在Seata-Raft模式中,client端在启动时会从配置中心获取当前client的事务分组(例如default)以及相关Raft集群节点的IP地址。通过向Seata-Server的控制端口发送请求,client可以获取到default分组对应的Raft集群的元数据,包括leader、follower和learner成员节点。然后,client会监视(watch)非leader节点的任意成员节点。

假设TM开始一个事务,并且本地的metadata中的leader节点指向了TC1的地址,那么TM只会与TC1进行交互。当TC1添加一个全局事务信息时,通过Raft协议,即图中标注为步骤1的日志发送,TC1会将日志发送给其他节点,步骤2是follower节点响应日志接收情况。当超过半数的节点(如TC2)接受并响应成功时,TC1上的状态机(FSM)将执行添加全局事务的动作。

watch watch2

如果TC1宕机或发生重选举,会发生什么呢?由于首次启动时已经获取到了元数据,client会执行watch follower节点的接口来更新本地的metadata信息。因此,后续的事务请求将发送到新的leader(例如TC2)。同时,TC1的数据已经被同步到了TC2和TC3,因此数据一致性不会受到影响。只在选举发生的瞬间,如果某个事务正好发送给了旧的leader,该事务会被主动回滚,以确保数据的正确性。

需要注意的是,在该模式下,如果事务处于决议发送请求或一阶段流程还未走完的时刻,并且恰好在选举时发生,这些事务会被主动回滚。因为RPC节点已经宕机或发生了重选举,当前没有实现RPC重试。TM侧默认有5次重试机制,但由于选举需要大约1s-2s的时间,这些处于begin状态的事务可能无法成功决议,因此会优先回滚,释放锁,以避免影响其他业务的正确性。

2.3.2 故障恢复

在Seata中,当TC发生故障时,数据恢复的过程如下:

故障恢复

如上图所示

  • 检查是否存在最新的数据快照:首先,系统会检查是否存在最新的数据快照文件。数据快照是基于内存的数据状态的一次全量拷贝,如果有最新的数据快照,则系统将直接加载该快照到内存中。

  • 根据快照后的Raft日志进行回放:如果存在最新的快照或者没有快照文件,系统将根据之前记录的Raft日志进行数据回放。每个Seata-Server中的请求最终会经过ServerOnRequestProcessor进行处理,然后转移到具体的协调者类(DefaultCoordinator或RaftCoordinator)中,再转向具体的业务代码(DefaultCore)进行相应的事务处理(如begin、commit、rollback等)。

  • 当日志回放完成后,便会由leader发起日志的同步,并继续执行相关事务的增删改动作。f

通过以上步骤,Seata能够实现在故障发生后的数据恢复。首先尝试加载最新的快照,如果有的话可以减少回放的时间;然后根据Raft日志进行回放,保证数据操作的一致性;最后通过日志同步机制,确保数据在多节点之间的一致性。

2.3.3 业务处理同步过程

流程 对于client侧获取最新metadata时恰好有业务线程在执行begin、commit或registry等操作的情况,Seata采取了以下处理方式:

  • client侧:

    • 如果客户端正在执行begin、commit或registry等操作,并且此时需要获取最新metadata,由于此时的leader可能已经不存在或不是当前leader,因此客户端的RPC请求可能会失败。
    • 如果请求失败,客户端会收到异常响应,此时客户端需要根据请求的结果进行回滚操作。
  • TC侧对旧leader的检测:

    • 在TC侧,如果此时客户端的请求到达旧的leader节点,TC会进行当前是否是leader的检测,如果不是leader,则会拒绝该请求。
    • 如果是leader但在中途失败,比如在提交任务到状态机的过程中失败,由于当前已经不是leader,创建任务(createTask)的动作会失败。这样,客户端也会接收到响应异常。
    • 旧leader的提交任务也会失败,确保了事务信息的一致性。 通过上述处理方式,当客户端获取最新metadata时恰好遇到业务操作的情况,Seata能够保证数据的一致性和事务的正确性。如果客户端的RPC请求失败,将触发回滚操作;而在TC侧,对旧leader的检测和任务提交的失败可以防止事务信息不一致的问题。这样,客户端的数据也能保持一致性。

3.使用部署

在使用和部署上,社区秉持着最小侵入,最小改动的原则,所以整体的部署上手应该是非常简单的,接下来分开client与server两端的部署改动点进行介绍

3.1 client

首先,使用注册配置中心较多的同学应该知道Seata的配置项中有一个seata.registry.type的配置项,支持了nacos,zk,etcd,redis等等,而在2.0以后增加了一个raft的配置项

   registry:
type: raft
raft:
server-addr: 192.168.0.111:7091, 192.168.0.112:7091, 192.168.0.113:7091

registry.type 改为raft,并配置raft相关元数据的获取地址,该地址统一为seata-server的ip+http端口 然后必不可少的就是传统的事务分组的配置

seata:
tx-service-group: default_tx_group
service:
vgroup-mapping:
default_tx_group: default

如现在使用的事务分组为default_tx_group,那么对应的seata集群/分组就是default,这个是有对应关系的,后续再server部署环节上会介绍 至此client的改动已经完成了

3.2 server

对于server的改动可能会多一些,要熟悉一些调优参数和配置,当然也可以选择默认值不做任何修改

seata:
server:
raft:
group: default #此值代表该raft集群的group,client的事务分组对应的值要与之对应
server-addr: 192.168.0.111:9091,192.168.0.112:9091,192.168.0.113:9091 # 3台节点的ip和端口,端口为该节点的netty端口+1000,默认netty端口为8091
snapshot-interval: 600 # 600秒做一次数据的快照,以便raftlog的快速滚动,但是每次做快照如果内存中事务数据过多会导致每600秒产生一次业务rt的抖动,但是对于故障恢复比较友好,重启节点较快,可以调整为30分钟,1小时都行,具体按业务来,可以自行压测看看是否有抖动,在rt抖动和故障恢复中自行找个平衡点
apply-batch: 32 # 最多批量32次动作做一次提交raftlog
max-append-bufferSize: 262144 #日志存储缓冲区最大大小,默认256K
max-replicator-inflight-msgs: 256 #在启用 pipeline 请求情况下,最大 in-flight 请求数,默认256
disruptor-buffer-size: 16384 #内部 disruptor buffer 大小,如果是写入吞吐量较高场景,需要适当调高该值,默认 16384
election-timeout-ms: 1000 #超过多久没有leader的心跳开始重选举
reporter-enabled: false # raft自身的监控是否开启
reporter-initial-delay: 60 # 监控的区间间隔
serialization: jackson # 序列化方式,不要改动
compressor: none # raftlog的压缩方式,如gzip,zstd等
sync: true # raft日志的刷盘方式,默认是同步刷盘
config:
# support: nacos, consul, apollo, zk, etcd3
type: file # 该配置可以选择不同的配置中心
registry:
# support: nacos, eureka, redis, zk, consul, etcd3, sofa
type: file # raft模式下不允许使用非file的其他注册中心
store:
# support: file 、 db 、 redis 、 raft
mode: raft # 使用raft存储模式
file:
dir: sessionStore # 该路径为raftlog及事务相关日志的存储位置,默认是相对路径,最好设置一个固定的位置

在3个或者大于3个节点的seata-server中配置完以上参数后,直接启动便可以看到类似以下的日志输出,就代表集群已经正常启动了

2023-10-13 17:20:06.392  WARN --- [Rpc-netty-server-worker-10-thread-1] [com.alipay.sofa.jraft.rpc.impl.BoltRaftRpcFactory] [ensurePipeline] []: JRaft SET bolt.rpc.dispatch-msg-list-in-default-executor to be false for replicator pipeline optimistic.
2023-10-13 17:20:06.439 INFO --- [default/PeerPair[10.58.16.231:9091 -> 10.58.12.217:9091]-AppendEntriesThread0] [com.alipay.sofa.jraft.storage.impl.LocalRaftMetaStorage] [save] []: Save raft meta, path=sessionStore/raft/9091/default/raft_meta, term=4, votedFor=0.0.0.0:0, cost time=25 ms
2023-10-13 17:20:06.441 WARN --- [default/PeerPair[10.58.16.231:9091 -> 10.58.12.217:9091]-AppendEntriesThread0] [com.alipay.sofa.jraft.core.NodeImpl] [handleAppendEntriesRequest] []: Node <default/10.58.16.231:9091> reject term_unmatched AppendEntriesRequest from 10.58.12.217:9091, term=4, prevLogIndex=4, prevLogTerm=4, localPrevLogTerm=0, lastLogIndex=0, entriesSize=0.
2023-10-13 17:20:06.442 INFO --- [JRaft-FSMCaller-Disruptor-0] [io.seata.server.cluster.raft.RaftStateMachine] [onStartFollowing] []: groupId: default, onStartFollowing: LeaderChangeContext [leaderId=10.58.12.217:9091, term=4, status=Status[ENEWLEADER<10011>: Raft node receives message from new leader with higher term.]].
2023-10-13 17:20:06.449 WARN --- [default/PeerPair[10.58.16.231:9091 -> 10.58.12.217:9091]-AppendEntriesThread0] [com.alipay.sofa.jraft.core.NodeImpl] [handleAppendEntriesRequest] []: Node <default/10.58.16.231:9091> reject term_unmatched AppendEntriesRequest from 10.58.12.217:9091, term=4, prevLogIndex=4, prevLogTerm=4, localPrevLogTerm=0, lastLogIndex=0, entriesSize=0.
2023-10-13 17:20:06.459 INFO --- [Bolt-default-executor-4-thread-1] [com.alipay.sofa.jraft.core.NodeImpl] [handleInstallSnapshot] []: Node <default/10.58.16.231:9091> received InstallSnapshotRequest from 10.58.12.217:9091, lastIncludedLogIndex=4, lastIncludedLogTerm=4, lastLogId=LogId [index=0, term=0].
2023-10-13 17:20:06.489 INFO --- [Bolt-conn-event-executor-13-thread-1] [com.alipay.sofa.jraft.rpc.impl.core.ClientServiceConnectionEventProcessor] [onEvent] []: Peer 10.58.12.217:9091 is connected
2023-10-13 17:20:06.519 INFO --- [JRaft-Group-Default-Executor-0] [com.alipay.sofa.jraft.util.Recyclers] [<clinit>] []: -Djraft.recyclers.maxCapacityPerThread: 4096.
2023-10-13 17:20:06.574 INFO --- [JRaft-Group-Default-Executor-0] [com.alipay.sofa.jraft.storage.snapshot.local.LocalSnapshotStorage] [destroySnapshot] []: Deleting snapshot sessionStore/raft/9091/default/snapshot/snapshot_4.
2023-10-13 17:20:06.574 INFO --- [JRaft-Group-Default-Executor-0] [com.alipay.sofa.jraft.storage.snapshot.local.LocalSnapshotStorage] [close] []: Renaming sessionStore/raft/9091/default/snapshot/temp to sessionStore/raft/9091/default/snapshot/snapshot_4.
2023-10-13 17:20:06.689 INFO --- [JRaft-FSMCaller-Disruptor-0] [io.seata.server.cluster.raft.snapshot.session.SessionSnapshotFile] [load] []: on snapshot load start index: 4
2023-10-13 17:20:06.694 INFO --- [JRaft-FSMCaller-Disruptor-0] [io.seata.server.cluster.raft.snapshot.session.SessionSnapshotFile] [load] []: on snapshot load end index: 4
2023-10-13 17:20:06.694 INFO --- [JRaft-FSMCaller-Disruptor-0] [io.seata.server.cluster.raft.RaftStateMachine] [onSnapshotLoad] []: groupId: default, onSnapshotLoad cost: 110 ms.
2023-10-13 17:20:06.694 INFO --- [JRaft-FSMCaller-Disruptor-0] [io.seata.server.cluster.raft.RaftStateMachine] [onConfigurationCommitted] []: groupId: default, onConfigurationCommitted: 10.58.12.165:9091,10.58.12.217:9091,10.58.16.231:9091.
2023-10-13 17:20:06.705 INFO --- [JRaft-FSMCaller-Disruptor-0] [com.alipay.sofa.jraft.storage.snapshot.SnapshotExecutorImpl] [onSnapshotLoadDone] []: Node <default/10.58.16.231:9091> onSnapshotLoadDone, last_included_index: 4
last_included_term: 4
peers: "10.58.12.165:9091"
peers: "10.58.12.217:9091"
peers: "10.58.16.231:9091"

2023-10-13 17:20:06.722 INFO --- [JRaft-Group-Default-Executor-1] [com.alipay.sofa.jraft.storage.impl.RocksDBLogStorage] [lambda$truncatePrefixInBackground$2] []: Truncated prefix logs in data path: sessionStore/raft/9091/default/log from log index 1 to 5, cost 0 ms.

3.3 faq

  • seata.raft.server-addr配置好后,必须通过server的openapi进行集群的扩缩容,直接改动该配置进行重启是不会生效的 接口为/metadata/v1/changeCluster?raftClusterStr=新的集群列表
  • 如果server-addr:中的地址都为本机,那么需要根据本机上不同的server的netty端口增加1000的偏移量,如server.port: 7092那么netty端口为8092,raft选举和通信端口便为9092,需要增加启动参数-Dserver.raftPort=9092. Linux下可以通过export JAVA_OPT="-Dserver.raftPort=9092"等方式指定。

4.压测对比

压测对比分为两种场景,并且为了避免数据热点冲突与线程调优等情况,将Client侧的数据初始化300W条商品,并直接使用jdk21虚拟线程+spring boot3+seata AT来测试,在gc方面全部采用分代ZGC进行,压测工具为阿里云PTS,Server侧统一使用jdk21(目前还未适配虚拟线程) 服务器配置如下 TC: 4c8g3 Client: 4c8G*1 数据库为阿里云rds 4c16g

  • 64并发压测只增加@Globaltransactional注解接口空提交的性能
  • 随机300W数据进行32并发10分钟的扣库存

4.1 1.7.1 db模式

raft压测模型

空提交 64C

db64-2

随机扣库存 32C

db32-2

4.2 2.0 raft模式

raft压测模型

空提交 64C

raft64-2

随机扣库存 32C

raft32c-2

4.3 压测结果对比

32并发对300W商品随机扣库存场景

tps avgtps maxcountrterror存储类型
1709(42%↑)2019(21%↑)1228803(42%↑)13.86ms(30%↓)0Raft
1201166886410519.86ms0DB

64并发空压@Globaltransactional接口(压测峰值上限为8000)

tps avgtps maxcountrterror存储类型
5704(20%↑)8062(30%↑)4101236(20%↑)7.79ms(19%↓)0Raft
4743617234102409.65ms0DB

除了以上数据上的直观对比,通过对压测的曲线图来观察,raft模式下tps与rt更加平稳,抖动更少,性能与吞吐量上更佳.

5.总结

在Seata未来的发展中,性能、入门门槛、部署运维成本,都是我们需要关注和不断优化的方向,在raft模式推出后有以下几个特点:

  1. 如存储方面,存算分离后Seata对其优化的上限被拔高,自主可控
  2. 部署成本更低,无需额外的注册中心,存储中间件
  3. 入门的门槛更低,无需学习其他的一些如注册中心的知识,一站式直接使用Seata Raft 即可上手

针对业界发展趋势,一些开源项目如ClickHouse和Kafka已经开始放弃使用ZooKeeper,并转而采用自研的解决方案,比如ClickKeeper和KRaft。这些方案将元数据等信息交由自身保证存储,以减少对第三方依赖的需求,从而降低运维成本和学习成本。这些特性是非常成熟和可借鉴的。

当然,目前来看,基于Raft模式的解决方案可能还不够成熟,可能无法完全达到上述描述的那样美好。然而,正是因为存在这样的理论基础,社区更应该朝着这个方向努力,让实践逐步接近理论的要求。在这里,欢迎所有对Seata感兴趣的同学加入社区,共同为Seata添砖加瓦!

· 阅读需 23 分钟

本文主要介绍分布式事务从内部到商业化和开源的演进历程,Seata社区当前进展和未来规划。

Seata是一款开源的分布式事务解决方案,旨在为现代化微服务架构下的分布式事务提供解决方案。Seata提供了完整的分布式事务解决方案,包括AT、TCC、Saga和XA事务模式,可支持多种编程语言和数据存储方案。Seata还提供了简便易用的API,以及丰富的文档和示例,方便企业在应用Seata时进行快速开发和部署。

Seata的优势在于具有高可用性、高性能、高扩展性等特点,同时在进行横向扩展时也无需做额外的复杂操作。 目前Seata已在阿里云上几千家客户业务系统中使用,其可靠性得到了业内各大厂商的认可和应用。

作为一个开源项目,Seata的社区也在不断扩大,现已成为开发者交流、分享和学习的重要平台,也得到了越来越多企业的支持和关注。

今天我主要针对以下三个小议题对Seata进行分享:

  • 从TXC/GTS 到 Seata
  • Seata 社区最新进展
  • Seata 社区未来规划

从TXC/GTS 到Seata

分布式事务的缘起

产品矩阵 Seata 在阿里内部的产品代号叫TXC(taobao transaction constructor),这个名字有非常浓厚的组织架构色彩。TXC 起源于阿里五彩石项目,五彩石是上古神话中女娲补天所用的石子,项目名喻意为打破关键技术壁垒,象征着阿里在从单体架构向分布式架构的演进过程中的重要里程碑。在这个项目的过程中演进出一批划时代的互联网中间件,包括我们常说的三大件:

  • HSF 服务调用框架
    解决单体应用到服务化后的服务通信调用问题。
  • TDDL 分库分表框架
    解决规模化后单库存储容量和连接数问题。
  • MetaQ 消息框架
    解决异步调用问题。

三大件的诞生满足了微服务化业务开发的基本需求,但是微服务化后的数据一致性问题并未得到妥善解决,缺少统一的解决方案。应用微服务化后出现数据一致性问题概率远大于单体应用,从进程内调用到网络调用这种复杂的环境加剧了异常场景的产生,服务跳数的增多使得在出现业务处理异常时无法协同上下游服务同时进行数据回滚。TXC的诞生正是为了解决应用架构层数据一致性的痛点问题,TXC 核心要解决的数据一致性场景包括:

  • 跨服务的一致性。 应对系统异常如调用超时和业务异常时协调上下游服务节点回滚。
  • 分库分表的数据一致性。 应对业务层逻辑SQL操作的数据在不同数据分片上,保证其分库分表操作的内部事务。
  • 消息发送的数据一致性。 应对数据操作和消息发送成功的不一致性问题。

为了克服以上通用场景遇到的问题,TXC与三大件做了无缝集成。业务使用三大件开发时,完全感知不到背后TXC的存在,业务不需要考虑数据一致性的设计问题,数据一致性保证交给了框架托管,业务更加聚焦于业务本身的开发,极大的提升了开发的效率。


GTS架构

TXC已在阿里集团内部广泛应用多年,经过双11等大型活动的洪荒流量洗礼,TXC极大提高了业务的开发效率,保证了数据的正确性,消除了数据不一致导致的资损和商誉问题。随着架构的不断演进,标准的三节点集群已可以承载接近10W TPS的峰值和毫秒级事务处理。在可用性和性能方面都达到了4个9的SLA保证,即使在无值守状态下也能保证全年无故障。


分布式事务的演进

新事物的诞生总是会伴随着质疑的声音。中间件层来保证数据一致性到底可靠吗?TXC最初的诞生只是一种模糊的理论,缺乏理论模型和工程实践。在我们进行MVP(最小可行产品)模型测试并推广业务上线后,经常出现故障,常常需要在深夜起床处理问题,睡觉时要佩戴手环来应对紧急响应,这也是我接管这个团队在技术上过的最痛苦的几年。

分布式事务演进

随后,我们进行了广泛的讨论和系统梳理。我们首先需要定义一致性问题,我们是要像RAFT一样实现多数共识一致性,还是要像Google Spanner一样解决数据库一致性问题,还是其他方式?从应用节点自上而下的分层结构来看,主要包括开发框架、服务调用框架、数据中间件、数据库Driver和数据库。我们需要决定在哪一层解决数据一致性问题。我们比较了解决不同层次数据一致性问题所面临的一致性要求、通用性、实现复杂度和业务接入成本。最后,我们权衡利弊,把实现复杂度留给我们,作为一个一致性组件,我们需要确保较高的一致性,但又不能锁定到具体数据库的实现上,确保场景的通用性和业务接入成本足够低以便更容易实现业务,这也是TXC最初采用AT模式的原因。

分布式事务它不仅仅是一个框架,它是一个体系。 我们在理论上定义了一致性问题,概念上抽象出了模式、角色、动作和隔离性等。从工程实践的角度,我们定义了编程模型,包括低侵入的注解、简单的方法模板和灵活的API ,定义了事务的基础能力和增强能力(例如如何以低成本支持大量活动),以及运维、安全、性能、可观测性和高可用等方面的能力。

事务逻辑模型 分布式事务解决了哪些问题呢?一个经典且具有体感的例子就是转账场景。转账过程包括减去余额和增加余额两个步骤,我们如何保证操作的原子性?在没有任何干预的情况下,这两个步骤可能会遇到各种问题,例如B账户已销户或出现服务调用超时等情况。

超时问题一直是分布式应用中比较难解决的问题,我们无法准确知晓B服务是否执行以及其执行顺序。从数据的角度来看,这意味着B 账户的钱未必会被成功加起来。在服务化改造之后,每个节点仅获知部分信息,而事务本身需要全局协调所有节点,因此需要一个拥有上帝视角、能够获取全部信息的中心化角色,这个角色就是TC(transaction coordinator),它用于全局协调事务的状态。TM(Transaction Manager) 则是驱动事务生成提议的角色。但是,即使上帝也有打瞌睡的时候,他的判断也并不总是正确的,因此需要一个RM(resource manager) 角色作为灵魂的代表来验证事务的真实性。这就是TXC 最基本的哲学模型。我们从方法论上验证了它的数据一致性是非常完备的,当然,我们的认知是有边界的。也许未来会证明我们是火鸡工程师,但在当前情况下,它的模型已经足以解决大部分现有问题。

分布式事务性能 经过多年的架构演进,从事务的单链路耗时角度来看,TXC在事务开始时的处理平均时间约为0.2毫秒,分支注册的平均时间约为0.4毫秒,整个事务额外的耗时在毫秒级别之内。这也是我们推算出的极限理论值。在吞吐量方面,单节点的TPS达到3万次/秒,标准集群的TPS接近10万次/秒。


Seata 开源

为什么要做开源?这是很多人问过我的问题。2017年我们做了商业化的 GTS(Global Transaction Service )产品产品在阿里云上售卖,有公有云和专有云两种形态。此时集团内发展的顺利,但是在我们商业化的过程中并不顺利,我们遇到了各种各样的问题,问题总结起来主要包括两类:一是开发者对于分布式事务的理论相当匮乏, 大多数人连本地事务都没搞明白是怎么回事更何况是分布式事务。 二是产品成熟度上存在问题, 经常遇到稀奇古怪的场景问题,导致了支持交付成本的急剧上升,研发变成了售后客服。

我们反思为什么遇到如此多的问题,这里主要的问题是在阿里集团内部是统一语言栈和统一技术栈的,我们对特定场景的打磨是非常成熟的,服务阿里一家公司和服务云上成千上万家企业有本质的区,这也启示我们产品的场景生态做的不够好。在GitHub 80%以上的开源软件是基础软件,基础软件首要解决的是场景通用性问题,因此它不能被有一家企业Lock In,比如像Linux,它有非常多的社区分发版本。因此,为了让我们的产品变得更好,我们选择了开源,与开发者们共建,普及更多的企业用户。

阿里开源 阿里的开源经历了三个主要阶段。第一个阶段是Dubbo所处的阶段,开发者用爱发电, Dubbo开源了有10几年的时间,时间充分证明了Dubbo是非常优秀的开源软件,它的微内核插件化的扩展性设计也是我最初开源Seata 的重要参考。做软件设计的时候我们要思考扩展性和性能权衡起来哪个会更重要一些,我们到底是要做一个三年的设计,五年的设计亦或是满足业务发展的十年设计。我们在做0-1服务调用问题的解决方案的同时,能否预测到1-100规模化后的治理问题。

第二个阶段是开源和商业化的闭环,商业化反哺于开源社区,促进了开源社区的发展。 我认为云厂商更容易做好开源的原因如下:

  • 首先,云是一个规模化的经济,必然要建立在稳定成熟的内核基础上,在上面去包装其产品化能力包括高可用、免运维和弹性能力。不稳定的内核必然导致过高的交付支持成本,研发团队的支持答疑穿透过高,过高的交付成本无法实现大规模的复制,穿透率过高无法使产品快速的演进迭代。
  • 其次,商业产品是更懂业务需求的。我们内部团队做技术的经常是站在研发的视角YY 需求,做出来的东西没有人使用,也就不会形成价值的转换。商业化收集到的都是真实的业务需求,因此,它的开源内核也必须会朝着这个方向演进。如果不朝着这个方向去演进必然导致两边架构上的分裂,增加团队的维护成本。
  • 最后,开源和商业化闭环,能促进双方更好的发展。如果开源内核经常出现各种问题,你是否愿意相信的它的商业化产品是足够优秀的。

第三个阶段是体系化和标准化。 首先,体系化是开源解决方案的基础。阿里的开源项目大多是基于内部电商场景的实践而诞生的。例如Higress,它用于打通蚂蚁集团的网关;Nacos承载着服务的百万实例和千万连接;Sentinel 提供大促时的降级和限流等高可用性能力;而Seata负责保障交易数据的一致性。这套体系化的开源解决方案是基于阿里电商生态的最佳实践而设计的。其次,标准化是另一个重要的特点。以OpenSergo为例,它既是一个标准,又是一个实现。在过去几年里,国内开源项目数量呈爆发式增长。然而,各个开源产品的能力差异很大,彼此集成时会遇到许多兼容性问题。因此,像OpenSergo这样的开源项目能够定义一些标准化的能力和接口,并提供一些实现,这将为整个开源生态系统的发展提供极大的帮助。


Seata 社区最新进展

Seata 社区简介

社区简介 目前,Seata已经开源了4种事务模式,包括AT、TCC、Saga和XA,并在积极探索其他可行的事务解决方案。 Seata已经与10多个主流的RPC框架和关系数据库进行了集成,同时与20 多个社区存在集成和被集成的关系。此外,我们还在多语言体系上探索除Java之外的语言,如Golang、PHP、Python和JS。

Seata已经被几千家客户应用到业务系统中。Seata的应用已经变得越来越成熟,在金融业务场景中信银行和光大银行与社区做了很好的合作,并成功将其纳入到核心账务系统中。在金融场景对微服务体系的落地是非常严苛的,这也标志着Seata的内核成熟度迈上了一个新台阶。


Seata 扩展生态

扩展生态 Seata采用了微内核和插件化的设计,它在API、注册配置中心、存储模式、锁控制、SQL解析器、负载均衡、传输、协议编解码、可观察性等方面暴露了丰富的扩展点。 这使得业务可以方便地进行灵活的扩展和技术组件的选择。


Seata 应用案例

应用案例 案例1:中航信航旅纵横项目
中航信航旅纵横项目在Seata 0.2版本中引入Seata解决机票和优惠券业务的数据一致性问题,大大提高了开发效率、减少了数据不一致造成的资损并提升了用户交互体验。

案例2:滴滴出行二轮车事业部
滴滴出行二轮车事业部在Seata 0.6.1版本中引入Seata,解决了小蓝单车、电动车、资产等业务流程的数据一致性问题,优化了用户使用体验并减少了资产的损失。

案例3:美团基础架构
美团基础架构团队基于开源的Seata项目开发了内部分布式事务解决方案Swan,被用于解决美团内部各业务的分布式事务问题。

场景4:盒马小镇
盒马小镇在游戏互动中使用Seata控制偷花的流程,开发周期大大缩短,从20天缩短到了5天,有效降低了开发成本。


Seata 事务模式的演进

模式演进


Seata 当前进展

  • 支持 Oracle和 Postgresql 多主键。
  • 支持 Dubbo3
  • 支持 Spring Boot3
  • 支持 JDK 17
  • 支持 ARM64 镜像
  • 支持多注册模型
  • 扩展了多种SQL语法
  • 支持 GraalVM Native Image
  • 支持 Redis lua 存储模式

Seata 2.x 发展规划

发展规划

主要包括下面几个方面:

  • 存储/协议/特性
    存储模式上探索存算不分离的Raft集群模式;更好的体验,统一当前4种事务模式的API;兼容GTS协议;支持Saga注解;支持分布式锁的控制;支持以数据视角的洞察和治理。
  • 生态
    融合支持更多的数据库,更多的服务框架,同时探索国产化信创生态的支持;支持MQ生态;进一步完善APM的支持。
  • 解决方案
    解决方案上除了支持微服务生态探索多云方案;更贴近云原生的解决方案;增加安全和流量防护能力;实现架构上核心组件的自闭环收敛。
  • 多语言生态
    多语言生态中Java最成熟,其他已支持的编程语言继续完善,同时探索与语言无关的Transaction Mesh方案。
  • 研发效能/体验
    提升测试的覆盖率,优先保证质量、兼容性和稳定性;重构官网文档结构,提升文档搜索的命中率;在体验上简化运维部署,实现一键安装和配置元数据简化;控制台支持事务控制和在线分析能力。

一句话总结2.x 的规划:更大的场景,更大的生态,从可用到好用。


Seata 社区联系方式

联系方式

· 阅读需 12 分钟

Seata简介

Seata的前身是阿里巴巴集团内大规模使用保证分布式事务一致性的中间件,Seata是其开源产品,由社区维护。在介绍Seata前,先与大家讨论下我们业务发展过程中经常遇到的一些问题场景。

业务场景

我们业务在发展的过程中,基本上都是从一个简单的应用,逐渐过渡到规模庞大、业务复杂的应用。这些复杂的场景难免遇到分布式事务管理问题,Seata的出现正是解决这些分布式场景下的事务管理问题。介绍下其中几个经典的场景:

场景一:分库分表场景下的分布式事务

image.png 起初我们的业务规模小、轻量化,单一数据库就能保障我们的数据链路。但随着业务规模不断扩大、业务不断复杂化,通常单一数据库在容量、性能上会遭遇瓶颈。通常的解决方案是向分库、分表的架构演进。此时,即引入了分库分表场景下的分布式事务场景。

场景二:跨服务场景下的分布式事务

image.png 降低单体应用复杂度的方案:应用微服务化拆分。拆分后,我们的产品由多个功能各异的微服务组件构成,每个微服务都使用独立的数据库资源。在涉及到跨服务调用的数据一致性场景时,就引入了跨服务场景下的分布式事务。

Seata架构

image.png 其核心组件主要如下:

  • Transaction Coordinator(TC)

事务协调器,维护全局事务的运行状态,负责协调并驱动全局事务的提交或回滚。

  • Transaction Manager(TM)

控制全局事务的边界,负责开启一个全局事务,并最终发起全局提交或全局回滚的决议,TM定义全局事务的边界。

  • Resource Manager(RM)

控制分支事务,负责分支注册、状态汇报,并接收事务协调器的指令,驱动分支(本地)事务的提交和回滚。RM负责定义分支事务的边界和行为。

Seata的可观测实践

为什么需要可观测?

  • 分布式事务消息链路较复杂

Seata在解决了用户易用性和分布式事务一致性这些问题的同时,需要多次TC与TM、RM之间的交互,尤其当微服务的链路变复杂时,Seata的交互链路也会呈正相关性增加。这种情况下,其实我们就需要引入可观测的能力来观察、分析事物链路。

  • 异常链路、故障排查难定位,性能优化无从下手

在排查Seata的异常事务链路时,传统的方法需要看日志,这样检索起来比较麻烦。在引入可观测能力后,帮助我们直观的分析链路,快速定位问题;为优化耗时的事务链路提供依据。

  • 可视化、数据可量化

可视化能力可让用户对事务执行情况有直观的感受;借助可量化的数据,可帮助用户评估资源消耗、规划预算。

可观测能力概览

可观测维度seata期望的能力技术选型参考
Metrics功能层面:可按业务分组隔离,采集事务总量、耗时等重要指标
性能层面:高度量性能,插件按需加载
架构层面:减少第三方依赖,服务端、客户端能够采用统一的架构,减少技术复杂度
兼容性层面:至少兼容Prometheus生态Prometheus:指标存储和查询等领域有着业界领先的地位
OpenTelemetry:可观测数据采集和规范的事实标准。但自身并不负责数据的存储,展示和分析
Tracing功能层面:全链路追踪分布式事务生命周期,反应分布式事务执行性能消耗
易用性方面:对使用seata的用户而言简单易接入SkyWalking:利用Java的Agent探针技术,效率高,简单易用。
Logging功能层面:记录服务端、客户端全部生命周期信息
易用性层面:能根据XID快速匹配全局事务对应链路日志Alibaba Cloud Service
ELK

Metrics维度

设计思路

  1. Seata作为一个被集成的数据一致性框架,Metrics模块将尽可能少的使用第三方依赖以降低发生冲突的风险
  2. Metrics模块将竭力争取更高的度量性能和更低的资源开销,尽可能降低开启后带来的副作用
  3. 配置时,Metrics是否激活、数据如何发布,取决于对应的配置;开启配置则自动启用,并默认将度量数据通过prometheusexporter的形式发布
  4. 不使用Spring,使用SPI(Service Provider Interface)加载扩展

模块设计

图片 1.png

  • seata-metrics-core:Metrics核心模块,根据配置组织(加载)1个Registry和N个Exporter;
  • seata-metrics-api:定义了Meter指标接口,Registry指标注册中心接口;
  • seata-metrics-exporter-prometheus:内置的prometheus-exporter实现;
  • seata-metrics-registry-compact:内置的Registry实现,并轻量级实现了Gauge、Counter、Summay、Timer指标;

metrics模块工作流

图片 1.png 上图是metrics模块的工作流,其工作流程如下:

  1. 利用SPI机制,根据配置加载Exporter和Registry的实现类;
  2. 基于消息订阅与通知机制,监听所有全局事务的状态变更事件,并publish到EventBus;
  3. 事件订阅者消费事件,并将生成的metrics写入Registry;
  4. 监控系统(如prometheus)从Exporter中拉取数据。

TC核心指标

image.png

TM核心指标

image.png

RM核心指标

image.png

大盘展示

lQLPJxZhZlqESU3NBpjNBp6w8zYK6VbMgzYCoKVrWEDWAA_1694_1688.png

Tracing维度

Seata为什么需要tracing?

  1. 对业务侧而言,引入Seata后,对业务性能会带来多大损耗?主要时间消耗在什么地方?如何针对性的优化业务逻辑?这些都是未知的。
  2. Seata的所有消息记录都通过日志持久化落盘,但对不了解Seata的用户而言,日志非常不友好。能否通过接入Tracing,提升事务链路排查效率?
  3. 对于新手用户,可通过Tracing记录,快速了解seata的工作原理,降低seata使用门槛。

Seata的tracing解决方案

  • Seata在自定义的RPC消息协议中定义了Header信息;
  • SkyWalking拦截指定的RPC消息,并注入tracing相关的span信息;
  • 以RPC消息的发出&接收为临界点,定义了span的生命周期范围。

基于上述的方式,Seata实现了事务全链路的tracing,具体接入可参考为[Seata应用 | Seata-server]接入Skywalking

tracing效果

  • 基于的demo场景:
  1. 用户请求交易服务
  2. 交易服务锁定库存
  3. 交易服务创建账单
  4. 账单服务进行扣款

image.png

  • GlobalCommit成功的事务链路(事例)

image.png image.png image.png

Logging维度

设计思路

image.png Logging这一块其实承担的是可观测这几个维度当中的兜底角色。放在最底层的,其实就是我们日志格式的设计,只有好日志格式,我们才能对它进行更好的采集、模块化的存储和展示。在其之上,是日志的采集、存储、监控、告警、数据可视化,这些模块更多的是有现成的工具,比如阿里的SLS日志服务、还有ELK的一套技术栈,我们更多是将开销成本、接入复杂度、生态繁荣度等作为考量。

日志格式设计

这里拿Seata-Server的一个日志格式作为案例: image.png

  • 线程池规范命名:当线程池、线程比较多时,规范的线程命名能将无序执行的线程执行次序清晰展示。
  • 方法全类名可追溯:快速定位到具体的代码块。
  • 重点运行时信息透出:重点突出关键日志,不关键的日志不打印,减少日志冗余。
  • 消息格式可扩展:通过扩展消息类的输出格式,减少日志的代码修改量。

总结&展望

Metrics

总结:基本实现分布式事务的可量化、可观测。 展望:更细粒度的指标、更广阔的生态兼容。

Tracing

总结:分布式事务全链路的可追溯。 展望:根据xid追溯事务链路,异常链路根因快速定位。

Logging

总结:结构化的日志格式。 展望:日志可观测体系演进。

· 阅读需 3 分钟

生产环境可用的 seata-go 1.2.0 来了

Seata 是一款开源的分布式事务解决方案,提供高性能和简单易用的分布式事务服务。

发布概览

Seata-go 1.2.0 版本支持 XA 模式。XA 协议是由 X/Open 组织提出的分布式事务处理规范,其优点是对业务代码无侵入。当前 Seata-go 的 XA 模式支持 MySQL 数据库。至此,seata-go 已经集齐 AT、TCC 和 XA 三种事务模式。 XA 模式的主要功能:

feature

  • [#467] 实现 XA 模式支持 MySQL
  • [#534] 支持 session 的负载均衡

bugfix

  • [#540] 修复初始化 xa 模式的 bug
  • [#545] 修复 xa 模式获取 db 版本号的 bug
  • [#548] 修复启动 xa 会失败的 bug
  • [#556] 修复 xa 数据源的 bug
  • [#562] 修复提交 xa 全局事务的 bug
  • [#564] 修复提交 xa 分支事务的 bug
  • [#566] 修复使用 xa 数据源执行本地事务的 bug

optimize

  • [#523] 优化 CI 流程
  • [#525] 将 jackson 序列化重命名为 json
  • [#532] 移除重复的代码
  • [#536] 优化 go import 代码格式
  • [#554] 优化 xa 模式的性能
  • [#561] 优化 xa 模式的日志输出

test

  • [#535] 添加集成测试

doc

  • [#550] 添加 1.2.0 版本的改动日志

contributors

Thanks to these contributors for their code commits. Please report an unintended omission.