多线程异步任务的事务协调与管理

为了在多线程环境下有效管理和协调异步任务的事务，我们设计了一个名为 `ThreadTransaction` 的类。该类利用了 Java 并发工具包（JUC）中的线程池以及 Spring 框架提供的事务管理功能，确保任务执行的原子性和一致性。

### 类结构及功能

- **日志记录**：通过 `LoggerFactory` 获取日志记录器，用于跟踪任务执行过程中的关键事件。
- **事务管理**：依赖于 `PlatformTransactionManager` 实现事务控制，确保每个任务的执行都在独立的事务中进行。
- **线程池配置**：支持两种类型的线程池——标准 JUC 线程池 (`ThreadPoolExecutor`) 和 Spring 提供的线程池 (`ThreadPoolTaskExecutor`)。
- **任务列表**：使用 `List` 存储待执行的任务，确保任务可以动态添加和管理。
- **执行计数器**：通过 `CountDownLatch` 实现对并发任务的同步控制，确保所有任务完成后才能进行最终的事务决策。
- **异常标记**：使用 `AtomicReference` 来追踪任务执行过程中是否有异常发生。

### 构造方法

`ThreadTransaction` 提供了多种构造方法，以适应不同的线程池配置需求：

1. 使用 `PlatformTransactionManager` 和 `ThreadPoolTaskExecutor` 初始化。
2. 使用 `PlatformTransactionManager` 和 `ThreadPoolExecutor` 初始化。
3. 支持指定初始任务数量的构造方法。

### 核心方法

- **addFunction(Supplier supplier)**：将需要异步执行的任务添加到任务列表中。
- **execute()**：启动所有异步任务并等待它们完成。根据异常标记字段的结果，决定是提交还是回滚事务。

#### 内部类 TransactionRunnable

`TransactionRunnable` 是一个实现了 `Runnable` 接口的内部类，负责具体任务的执行和事务管理。它会在每个任务开始时创建一个新的事务，并在任务完成后根据全局异常标志决定事务的提交或回滚。

```java
public class ThreadTransaction {
private final Logger LOG = LoggerFactory.getLogger(ThreadTransaction.class);
private PlatformTransactionManager platformTransactionManager;
private ThreadPoolExecutor threadPoolExecutor;
private ThreadPoolTaskExecutor threadPoolTaskExecutor;
private List supplierList = new ArrayList<>();
private volatile CountDownLatch countDownLatch;
private AtomicReference isError = new AtomicReference<>(false);

public ThreadTransaction(PlatformTransactionManager platformTransactionManager, ThreadPoolTaskExecutor threadPoolTaskExecutor) {
this.platformTransactiOnManager= platformTransactionManager;
this.threadPoolTaskExecutor = threadPoolTaskExecutor;
}

public ThreadTransaction(PlatformTransactionManager platformTransactionManager, ThreadPoolExecutor threadPoolExecutor) {
this.platformTransactiOnManager= platformTransactionManager;
this.threadPoolExecutor = threadPoolExecutor;
}

public ThreadTransaction(PlatformTransactionManager platformTransactionManager, ThreadPoolTaskExecutor threadPoolTaskExecutor, int size) {
this(platformTransactionManager, threadPoolTaskExecutor);
supplierList = new ArrayList<>(size);
}

public boolean addFunction(Supplier supplier) {
return supplierList.add(supplier);
}

public void execute() {
LOG.info("多线程事务开始...");
countDownLatch = new CountDownLatch(supplierList.size());
for (Supplier supplier : supplierList) {
this.threadPoolTaskExecutor.submit(new TransactionRunnable(platformTransactionManager, supplier));
}
try {
countDownLatch.await();
if (isError.get()) {
LOG.error("多线程执行失败，事务已回滚！");
throw new RuntimeException("多线程执行失败！");
}
LOG.info("多线程执行成功，事务已提交！");
} catch (Exception e) {
LOG.error("多线程执行失败：" + e.getMessage());
e.printStackTrace();
}
}

class TransactionRunnable implements Runnable {
private PlatformTransactionManager platformTransactionManager;
private Supplier supplier;

public TransactionRunnable(PlatformTransactionManager platformTransactionManager, Supplier supplier) {
this.platformTransactiOnManager= platformTransactionManager;
this.supplier = supplier;
}

@Override
public void run() {
DefaultTransactionDefinition defaultTransactiOnDefinition= new DefaultTransactionDefinition();
defaultTransactionDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
TransactionStatus transaction = this.platformTransactionManager.getTransaction(defaultTransactionDefinition);
try {
this.supplier.get();
} catch (Exception e) {
isError.set(true);
LOG.error("多线程事务执行失败: {}", e.getMessage());
e.printStackTrace();
}
countDownLatch.countDown();
try {
if (isError.get()) {
LOG.info("多线程事务（子线程）回滚");
platformTransactionManager.rollback(transaction);
} else {
LOG.info("多线程事务（子线程）提交");
platformTransactionManager.commit(transaction);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
```

通过这种方式，`ThreadTransaction` 类能够在多线程环境下有效地管理异步任务的事务，确保数据一致性和系统的稳定性。