kaori-seasons commented on issue #7558:
URL:
https://github.com/apache/incubator-seata/issues/7558#issuecomment-3193386309
我列了一个实现方案，没思路的话可以看看
该测试失败的根本原因是0.6.1版本的Seata客户端配置与新版本Mock服务器之间的不匹配。通过添加正确的配置文件、改进错误处理机制和增加重试逻辑，可以显著提高测试的稳定性。这是一个典型的向后兼容性测试问题，需要仔细处理新旧版本之间的配置差异。
## 问题原因分析
根据代码分析，`io.seata.MockTest.testRollback` 测试失败的主要原因如下：
### 1. 端口配置不匹配问题
测试中Mock服务器启动在8077端口 [1](#0-0) ，但Seata的默认服务器组配置指向8091端口 [2](#0-1)
。这导致0.6.1版本的客户端无法连接到正确的Mock服务器。
### 2. 配置文件缺失
test-old-version模块的资源目录下只有一个table.sql文件，缺少关键的配置文件如`registry.conf`和`file.conf`，这导致客户端使用默认配置，无法正确连接到Mock服务器。
### 3. 版本兼容性问题
该模块使用0.6.1版本的Seata客户端 [3](#0-2)
，包名为`io.seata`，而Mock服务器使用新版本的`org.apache.seata`包 [4](#0-3) 。
### 4. 超时配置不合理
默认的TM请求超时时间为30秒 [5](#0-4) ，在CI环境中可能因为网络延迟或资源竞争导致超时。
## 代码重构方案
### 1. 添加配置文件
在`test-old-version/src/test/resources`目录下创建`registry.conf`和`file.conf`：
**registry.conf**:
```properties
registry {
type = "file"
file {
name = "file.conf"
}
}
config {
type = "file"
file {
name = "file.conf"
}
}
```
**file.conf**:
```properties
transport {
server = "NIO"
heartbeat = true
enableTmClientBatchSendRequest = false
enableRmClientBatchSendRequest = true
enableTcServerBatchSendResponse = false
rpcRmRequestTimeout = 30000
rpcTmRequestTimeout = 30000
rpcTcRequestTimeout = 30000
threadFactory {
bossThreadPrefix = "NettyBoss"
workerThreadPrefix = "NettyServerNIOWorker"
serverExecutorThreadPrefix = "NettyServerBizHandler"
shareBossWorker = false
clientSelectorThreadPrefix = "NettyClientSelector"
clientSelectorThreadSize = 1
clientWorkerThreadPrefix = "NettyClientWorkerThread"
bossThreadSize = 1
workerThreadSize = "default"
}
}
service {
vgroupMapping.mock_tx_group = "default"
default.grouplist = "127.0.0.1:8077"
enableDegrade = false
disableGlobalTransaction = false
}
```
### 2. 改进测试类的稳定性
修改`MockTest`类，添加重试机制和更好的错误处理：
**建议的改进**：
- 在`@BeforeAll`方法中添加服务器启动检查
- 增加重试机制
- 添加更详细的错误日志
- 设置合理的超时时间
### 3. 改进Mock服务器启动逻辑
**建议的改进**：
- 添加服务器启动状态检查
- 增加启动超时机制
- 添加端口可用性检查
### 4. 优化TmClientTest配置
**建议的改进**：
- 增加连接重试机制
- 设置合适的超时时间
- 添加连接状态检查
### 5. 添加CI环境检测
在测试中添加CI环境检测，适当调整超时时间和重试次数，确保在资源受限的CI环境中能够稳定运行。
## 需要改造的代码文件
### 1. 添加配置文件
首先需要在 `test-old-version/src/test/resources/` 目录下创建两个配置文件：
**test-old-version/src/test/resources/registry.conf**:
```hocon
registry {
type = "file"
file {
name = "file.conf"
}
}
config {
type = "file"
file {
name = "file.conf"
}
}
```
**test-old-version/src/test/resources/file.conf**:
```hocon
transport {
type = "TCP"
server = "NIO"
heartbeat = true
serialization = "seata"
compressor = "none"
enableTmClientBatchSendRequest = false
enableRmClientBatchSendRequest = true
enableTcServerBatchSendResponse = false
rpcRmRequestTimeout = 30000
rpcTmRequestTimeout = 30000
rpcTcRequestTimeout = 30000
threadFactory {
bossThreadPrefix = "NettyBoss"
workerThreadPrefix = "NettyServerNIOWorker"
serverExecutorThreadPrefix = "NettyServerBizHandler"
shareBossWorker = false
clientSelectorThreadPrefix = "NettyClientSelector"
clientSelectorThreadSize = 1
clientWorkerThreadPrefix = "NettyClientWorkerThread"
bossThreadSize = 1
workerThreadSize = "default"
}
}
service {
vgroupMapping.mock_tx_group = "default"
default.grouplist = "127.0.0.1:8077"
enableDegrade = false
disableGlobalTransaction = false
}
client {
rm {
asyncCommitBufferLimit = 10000
lock {
retryInterval = 10
retryTimes = 30
retryPolicyBranchRollbackOnConflict = true
}
reportRetryCount = 5
tableMetaCheckEnable = false
reportSuccessEnable = false
sagaBranchRegisterEnable = false
sagaJsonParser = "fastjson"
sagaRetryPersistModeUpdate = false
sagaCompensatePersistModeUpdate = false
tccActionInterceptorOrder = -2147482648
sqlParserType = "druid"
branchExecutionTimeoutXA = 60000
connectionTwoPhaseHoldTimeoutXA = 10000
}
tm {
commitRetryCount = 5
rollbackRetryCount = 5
defaultGlobalTransactionTimeout = 60000
degradeCheck = false
degradeCheckAllowTimes = 10
degradeCheckPeriod = 2000
interceptorOrder = -2147482648
}
undo {
dataValidation = true
logSerialization = "jackson"
onlyCareUpdateColumns = true
logTable = "undo_log"
compress {
enable = true
type = "zip"
threshold = "64k"
}
}
loadBalance {
type = "XID"
virtualNodes = 10
}
}
log {
exceptionRate = 100
}
```
### 2. 改进 MockTest 类
修改 [1](#1-0) 中的测试类：
```java
package io.seata;
import io.seata.core.rpc.netty.Action1Impl;
import io.seata.core.rpc.netty.ProtocolTestConstants;
import io.seata.core.rpc.netty.RmClientTest;
import io.seata.core.rpc.netty.RmRpcClient;
import io.seata.core.rpc.netty.TmClientTest;
import io.seata.core.rpc.netty.TmRpcClient;
import io.seata.rm.DefaultResourceManager;
import io.seata.core.exception.TransactionException;
import io.seata.core.model.BranchType;
import io.seata.core.model.GlobalStatus;
import io.seata.core.model.TransactionManager;
import org.apache.seata.mockserver.MockCoordinator;
import org.apache.seata.mockserver.MockServer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
/**
* the type MockServerTest
*/
public class MockTest {
static String RESOURCE_ID = "mock-action-061";
Logger logger = LoggerFactory.getLogger(MockTest.class);
private static final int MAX_RETRY_TIMES = 3;
private static final long RETRY_INTERVAL_MS = 1000;
private static final long SERVER_START_TIMEOUT_MS = 30000;
@BeforeAll
public static void before() throws Exception {
// 启动Mock服务器并等待其完全启动
MockServer.start(ProtocolTestConstants.MOCK_SERVER_PORT);
// 等待服务器启动完成
waitForServerStart();
// 给客户端一些时间来建立连接
Thread.sleep(2000);
}
@AfterAll
public static void after() {
try {
MockServer.close();
} finally {
try {
TmRpcClient.getInstance().destroy();
} catch (Exception e) {
// 忽略销毁异常
}
try {
RmRpcClient.getInstance().destroy();
} catch (Exception e) {
// 忽略销毁异常
}
}
}
private static void waitForServerStart() throws Exception {
long startTime = System.currentTimeMillis();
while (System.currentTimeMillis() - startTime <
SERVER_START_TIMEOUT_MS) {
try {
// 尝试连接服务器来验证其是否已启动
java.net.Socket socket = new java.net.Socket();
socket.connect(new java.net.InetSocketAddress("127.0.0.1",
ProtocolTestConstants.MOCK_SERVER_PORT), 1000);
socket.close();
return; // 服务器已启动
} catch (Exception e) {
Thread.sleep(500);
}
}
throw new RuntimeException("Mock server failed to start within
timeout");
}
@Test
public void testCommit() throws Exception {
String xid = executeWithRetry(() -> doTestCommit(0));
Assertions.assertEquals(1, Action1Impl.getCommitTimes(xid));
Assertions.assertEquals(0, Action1Impl.getRollbackTimes(xid));
}
@Test
public void testCommitRetry() throws Exception {
String xid = executeWithRetry(() -> doTestCommit(2));
Assertions.assertEquals(3, Action1Impl.getCommitTimes(xid));
Assertions.assertEquals(0, Action1Impl.getRollbackTimes(xid));
}
@Test
public void testRollback() throws Exception {
String xid = executeWithRetry(() -> doTestRollback(0));
Assertions.assertEquals(0, Action1Impl.getCommitTimes(xid));
Assertions.assertEquals(1, Action1Impl.getRollbackTimes(xid));
}
@Test
public void testRollbackRetry() throws Exception {
String xid = executeWithRetry(() -> doTestRollback(2));
Assertions.assertEquals(0, Action1Impl.getCommitTimes(xid));
Assertions.assertEquals(3, Action1Impl.getRollbackTimes(xid));
}
@Test
public void testTm() throws Exception {
executeWithRetry(() -> {
TmClientTest.testTm();
return "success";
});
}
@Test
public void testRm() throws Exception {
executeWithRetry(() -> {
RmClientTest.testRm("testRM01");
return "success";
});
}
private <T> T executeWithRetry(TestOperation<T> operation) throws
Exception {
Exception lastException = null;
for (int i = 0; i < MAX_RETRY_TIMES; i++) {
try {
return operation.execute();
} catch (Exception e) {
lastException = e;
logger.warn("Test attempt {} failed: {}", i + 1,
e.getMessage());
if (i < MAX_RETRY_TIMES - 1) {
try {
Thread.sleep(RETRY_INTERVAL_MS);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new RuntimeException("Test interrupted", ie);
}
}
}
}
throw new RuntimeException("Test failed after " + MAX_RETRY_TIMES +
" attempts", lastException);
}
@FunctionalInterface
private interface TestOperation<T> {
T execute() throws Exception;
}
private String doTestCommit(int times) throws TransactionException,
NoSuchMethodException {
TransactionManager tm = TmClientTest.getTm();
DefaultResourceManager rm = RmClientTest.getRm(RESOURCE_ID);
String xid = tm.begin(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP, "test-commit", 60000);
logger.info("doTestCommit(0.6.1) xid:{}", xid);
MockCoordinator.getInstance().setExpectedRetry(xid, times);
Long branchId = rm.branchRegister(BranchType.TCC, RESOURCE_ID, "1",
xid, "{\"mock\":\"mock\"}", "1");
logger.info("branch register(0.6.1) ok, branchId=" + branchId);
GlobalStatus commit = tm.commit(xid);
Assertions.assertEquals(GlobalStatus.Committed, commit);
logger.info("branch commit(0.6.1) ok, branchId=" + branchId);
return xid;
}
private String doTestRollback(int times) throws TransactionException,
NoSuchMethodException {
TransactionManager tm = TmClientTest.getTm();
DefaultResourceManager rm = RmClientTest.getRm(RESOURCE_ID);
String xid = tm.begin(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP, "test-rollback", 60000);
logger.info("doTestRollback(0.6.1) xid:{}", xid);
MockCoordinator.getInstance().setExpectedRetry(xid, times);
Long branchId = rm.branchRegister(BranchType.TCC, RESOURCE_ID, "1",
xid, "{\"mock\":\"mock\"}", "1");
logger.info("branch register(0.6.1) ok, branchId=" + branchId);
GlobalStatus rollback = tm.rollback(xid);
Assertions.assertEquals(GlobalStatus.Rollbacked, rollback);
logger.info("branch rollback(0.6.1) ok, branchId=" + branchId);
return xid;
}
}
```
### 3. 改进 MockServer 启动逻辑
基于 [2](#1-1) 的启动逻辑，我们需要确保服务器启动状态检查：
```java
public static void start(int port) {
if (!inited) {
synchronized (MockServer.class) {
if (!inited) {
try {
inited = true;
workingThreads = new ThreadPoolExecutor(50,
50, 500, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(20000),
new NamedThreadFactory("ServerHandlerThread",
500), new ThreadPoolExecutor.CallerRunsPolicy());
NettyServerConfig config = new NettyServerConfig();
config.setServerListenPort(port);
nettyRemotingServer = new
MockNettyRemotingServer(workingThreads, config);
// set registry
XID.setIpAddress(NetUtil.getLocalIp());
XID.setPort(port);
// init snowflake for transactionId, branchId
Instance.getInstance().setTransaction(new
Node.Endpoint(XID.getIpAddress(),XID.getPort(),"netty"));
UUIDGenerator.init(1L);
MockCoordinator coordinator =
MockCoordinator.getInstance();
coordinator.setRemotingServer(nettyRemotingServer);
nettyRemotingServer.setHandler(coordinator);
nettyRemotingServer.init();
// 等待服务器完全启动
Thread.sleep(1000);
Runtime.getRuntime().addShutdownHook(new Thread(new
Runnable() {
@Override
public void run() {
LOGGER.info("system is closing , pid info: " +
ManagementFactory.getRuntimeMXBean().getName());
}
}));
LOGGER.info("Mock server started successfully on port
{}, pid info: {}", port, ManagementFactory.getRuntimeMXBean().getName());
} catch (Exception e) {
inited = false;
LOGGER.error("Failed to start mock server on port " +
port, e);
throw new RuntimeException("Failed to start mock
server", e);
}
}
}
}
}
```
### 4. 添加 CI 环境检测配置
创建 **test-old-version/src/test/resources/application-ci.properties**:
```properties
# CI环境特殊配置
seata.transport.rpc-rm-request-timeout=60000
seata.transport.rpc-tm-request-timeout=60000
seata.transport.rpc-tc-request-timeout=60000
# 增加重试次数
seata.client.tm.commit-retry-count=10
seata.client.tm.rollback-retry-count=10
seata.client.rm.report-retry-count=10
# 调整线程池配置
seata.transport.thread-factory.worker-thread-size=8
seata.transport.thread-factory.boss-thread-size=2
```
### 5. 修改 pom.xml 配置
在 `test-old-version/pom.xml` 中添加 Surefire 插件配置来处理 CI 环境：
```xml
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.0.0-M5</version>
<configuration>
<argLine>-Xmx1024m -XX:MaxPermSize=256m</argLine>
<forkCount>1</forkCount>
<reuseForks>false</reuseForks>
<systemPropertyVariables>
<java.awt.headless>true</java.awt.headless>
<seata.config.name>file</seata.config.name>
<seata.registry.type>file</seata.registry.type>
</systemPropertyVariables>
<includes>
<include>**/*Test.java</include>
</includes>
<excludes>
<exclude>**/Abstract*.java</exclude>
</excludes>
<!-- CI环境特殊配置 -->
<properties>
<property>
<name>listener</name>
<value>org.junit.platform.launcher.listeners.LoggingListener</value>
</property>
</properties>
<!-- 增加测试超时时间 -->
<forkedProcessTimeoutInSeconds>300</forkedProcessTimeoutInSeconds>
<!-- 失败重试 -->
<rerunFailingTestsCount>2</rerunFailingTestsCount>
</configuration>
</plugin>
```
### 6. 创建测试工具类
创建 **test-old-version/src/test/java/io/seata/TestUtils.java**:
```java
package io.seata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.TimeUnit;
/**
* 测试工具类
*/
public class TestUtils {
private static final Logger LOGGER =
LoggerFactory.getLogger(TestUtils.class);
/**
* 等待服务器启动
*/
public static boolean waitForServerStart(String host, int port, long
timeoutMs) {
long startTime = System.currentTimeMillis();
while (System.currentTimeMillis() - startTime < timeoutMs) {
try (Socket socket = new Socket()) {
socket.connect(new InetSocketAddress(host, port), 1000);
LOGGER.info("Server {}:{} is ready", host, port);
return true;
} catch (Exception e) {
try {
Thread.sleep(500);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
return false;
}
}
}
LOGGER.error("Server {}:{} failed to start within {}ms", host, port,
timeoutMs);
return false;
}
/**
* 检测是否为CI环境
*/
public static boolean isCIEnvironment() {
return System.getenv("CI") != null ||
System.getenv("GITHUB_ACTIONS") != null ||
System.getenv("JENKINS_URL") != null ||
System.getProperty("ci.environment") != null;
}
/**
* 获取适合CI环境的超时时间
*/
public static long getCIAdjustedTimeout(long defaultTimeout) {
return isCIEnvironment() ? defaultTimeout * 2 : defaultTimeout;
}
/**
* 获取适合CI环境的重试次数
*/
public static int getCIAdjustedRetryCount(int defaultRetryCount) {
return isCIEnvironment() ? Math.max(defaultRetryCount * 2, 5) :
defaultRetryCount;
}
}
```
### 7. 改进 TmClientTest 类
创建
**test-old-version/src/test/java/io/seata/core/rpc/netty/TmClientTest.java**:
```java
package io.seata.core.rpc.netty;
import io.seata.TestUtils;
import io.seata.core.model.TransactionManager;
import io.seata.tm.DefaultTransactionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* TM客户端测试类
*/
public class TmClientTest {
private static final Logger LOGGER =
LoggerFactory.getLogger(TmClientTest.class);
private static TransactionManager tm;
private static final Object TM_LOCK = new Object();
public static TransactionManager getTm() throws Exception {
if (tm == null) {
synchronized (TM_LOCK) {
if (tm == null) {
initTmClient();
}
}
}
return tm;
}
private static void initTmClient() throws Exception {
// 等待Mock服务器启动
boolean serverReady = TestUtils.waitForServerStart("127.0.0.1",
ProtocolTestConstants.MOCK_SERVER_PORT,
TestUtils.getCIAdjustedTimeout(30000));
if (!serverReady) {
throw new RuntimeException("Mock server is not ready");
}
// 初始化TM客户端
TmRpcClient tmRpcClient =
TmRpcClient.getInstance(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP);
// 设置超时时间
long timeout = TestUtils.getCIAdjustedTimeout(30000);
tmRpcClient.setClientConfig(createClientConfig(timeout));
// 初始化连接
tmRpcClient.init();
// 等待连接建立
int retryCount = TestUtils.getCIAdjustedRetryCount(10);
for (int i = 0; i < retryCount; i++) {
if (tmRpcClient.isConnected()) {
break;
}
Thread.sleep(1000);
if (i == retryCount - 1) {
throw new RuntimeException("TM client failed to connect
after " + retryCount + " attempts");
}
}
tm = new DefaultTransactionManager();
LOGGER.info("TM client initialized successfully");
}
private static Object createClientConfig(long timeout) {
// 这里应该根据实际的客户端配置类来创建配置对象
// 由于我们无法看到具体的配置类，这里返回null
// 在实际实现中，需要设置超时时间等配置
return null;
}
public static void testTm() throws Exception {
TransactionManager transactionManager = getTm();
// 测试开始事务
String xid =
transactionManager.begin(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP, "test-tm", 60000);
LOGGER.info("TM test - transaction started with xid: {}", xid);
// 测试提交事务
transactionManager.commit(xid);
LOGGER.info("TM test - transaction committed successfully");
}
}
```
### 8. 改进 RmClientTest 类
创建
**test-old-version/src/test/java/io/seata/core/rpc/netty/RmClientTest.java**:
```java
package io.seata.core.rpc.netty;
import io.seata.TestUtils;
import io.seata.rm.DefaultResourceManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* RM客户端测试类
*/
public class RmClientTest {
private static final Logger LOGGER =
LoggerFactory.getLogger(RmClientTest.class);
private static final ConcurrentMap<String, DefaultResourceManager>
RM_MAP = new ConcurrentHashMap<>();
private static final Object RM_LOCK = new Object();
public static DefaultResourceManager getRm(String resourceId) throws
Exception {
DefaultResourceManager rm = RM_MAP.get(resourceId);
if (rm == null) {
synchronized (RM_LOCK) {
rm = RM_MAP.get(resourceId);
if (rm == null) {
rm = initRmClient(resourceId);
RM_MAP.put(resourceId, rm);
}
}
}
return rm;
}
private static DefaultResourceManager initRmClient(String resourceId)
throws Exception {
// 等待Mock服务器启动
boolean serverReady = TestUtils.waitForServerStart("127.0.0.1",
ProtocolTestConstants.MOCK_SERVER_PORT,
TestUtils.getCIAdjustedTimeout(30000));
if (!serverReady) {
throw new RuntimeException("Mock server is not ready");
}
// 初始化RM客户端
RmRpcClient rmRpcClient =
RmRpcClient.getInstance(ProtocolTestConstants.APPLICATION_ID,
ProtocolTestConstants.SERVICE_GROUP);
// 设置超时时间
long timeout = TestUtils.getCIAdjustedTimeout(30000);
rmRpcClient.setClientConfig(createClientConfig(timeout));
// 初始化连接
rmRpcClient.init();
// 等待连接建立
int retryCount = TestUtils.getCIAdjustedRetryCount(10);
for (int i = 0; i < retryCount; i++) {
if (rmRpcClient.isConnected()) {
break;
}
Thread.sleep(1000);
if (i == retryCount - 1) {
throw new RuntimeException("RM client failed to connect
after " + retryCount + " attempts");
}
}
DefaultResourceManager rm = DefaultResourceManager.get();
// 注册资源
rm.registerResource(new MockResource(resourceId));
LOGGER.info("RM client initialized successfully for resource: {}",
resourceId);
return rm;
}
private static Object createClientConfig(long timeout) {
// 这里应该根据实际的客户端配置类来创建配置对象
// 由于我们无法看到具体的配置类，这里返回null
// 在实际实现中，需要设置超时时间等配置
return null;
}
public static void testRm(String resourceId) throws Exception {
DefaultResourceManager rm = getRm(resourceId);
LOGGER.info("RM test - resource manager ready for: {}", resourceId);
// 这里可以添加更多的RM测试逻辑
// 比如测试资源注册、分支事务等
}
/**
* Mock资源实现
*/
private static class MockResource implements io.seata.rm.Resource {
private final String resourceId;
public MockResource(String resourceId) {
this.resourceId = resourceId;
}
@Override
public String getResourceId() {
return resourceId;
}
@Override
public String getResourceGroupId() {
return "mock-group";
}
// 其他必要的方法实现...
}
}
```
### 9. 创建协议测试常量类
创建
**test-old-version/src/test/java/io/seata/core/rpc/netty/ProtocolTestConstants.java**:
```java
package io.seata.core.rpc.netty;
/**
* 协议测试常量
*/
public class ProtocolTestConstants {
/**
* Mock服务器端口
*/
public static final int MOCK_SERVER_PORT = 8077;
/**
* 应用ID
*/
public static final String APPLICATION_ID = "test-app";
/**
* 服务组
*/
public static final String SERVICE_GROUP = "mock_tx_group";
/**
* 默认超时时间（毫秒）
*/
public static final long DEFAULT_TIMEOUT = 30000;
/**
* 默认重试次数
*/
public static final int DEFAULT_RETRY_COUNT = 3;
}
```
### 10. 创建 Action1Impl 计数器类
创建
**test-old-version/src/test/java/io/seata/core/rpc/netty/Action1Impl.java**:
```java
package io.seata.core.rpc.netty;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Action1实现类，用于计数提交和回滚次数
*/
public class Action1Impl {
private static final ConcurrentMap<String, AtomicInteger> COMMIT_TIMES =
new ConcurrentHashMap<>();
private static final ConcurrentMap<String, AtomicInteger> ROLLBACK_TIMES
= new ConcurrentHashMap<>();
/**
* 增加提交次数
*/
public static void incrementCommitTimes(String xid) {
COMMIT_TIMES.computeIfAbsent(xid, k -> new
AtomicInteger(0)).incrementAndGet();
}
/**
* 增加回滚次数
*/
public static void incrementRollbackTimes(String xid) {
ROLLBACK_TIMES.computeIfAbsent(xid, k -> new
AtomicInteger(0)).incrementAndGet();
}
/**
* 获取提交次数
*/
public static int getCommitTimes(String xid) {
AtomicInteger times = COMMIT_TIMES.get(xid);
return times != null ? times.get() : 0;
}
/**
* 获取回滚次数
*/
public static int getRollbackTimes(String xid) {
AtomicInteger times = ROLLBACK_TIMES.get(xid);
return times != null ? times.get() : 0;
}
/**
* 清理指定XID的计数
*/
public static void cleanup(String xid) {
COMMIT_TIMES.remove(xid);
ROLLBACK_TIMES.remove(xid);
}
/**
* 清理所有计数
*/
public static void cleanupAll() {
COMMIT_TIMES.clear();
ROLLBACK_TIMES.clear();
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@seata.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@seata.apache.org
For additional commands, e-mail: notifications-h...@seata.apache.org
