Spring Batch作业在虚拟线程StepTaskExecutor中事务隔离级别传播失效？PlatformTransactionManager与ThreadLocal事务上下文

Spring Batch 与虚拟线程：事务隔离的挑战

大家好，今天我们来探讨一个Spring Batch中比较棘手的问题：在使用虚拟线程（Virtual Threads）作为 StepTaskExecutor 时，事务隔离级别传播可能失效的情况。这个问题涉及到PlatformTransactionManager、ThreadLocal事务上下文以及虚拟线程的特性，理解起来需要一定的基础，但对我们编写健壮的Spring Batch应用至关重要。

1. 事务隔离级别回顾

在深入问题之前，我们先简单回顾一下事务隔离级别。事务隔离级别定义了并发事务之间的可见性程度，以及它们之间可能产生的干扰。常见的隔离级别包括：

隔离级别	描述	可能出现的问题
READ_UNCOMMITTED	允许读取未提交的数据。	脏读（Dirty Reads）：读取到其他事务尚未提交的数据。
READ_COMMITTED	只能读取已提交的数据。	不可重复读（Non-Repeatable Reads）：在同一事务中，多次读取同一数据，由于其他事务的提交，导致每次读取的结果不一致。
REPEATABLE_READ	保证在同一事务中，多次读取同一数据，结果始终一致。	幻读（Phantom Reads）：在同一事务中，多次执行查询，由于其他事务的插入或删除操作，导致每次查询返回的记录数不一致（主要针对范围查询）。
SERIALIZABLE	最高级别的隔离，强制事务串行执行，避免并发问题。	并发性能低。

Spring Batch 默认使用数据库的默认隔离级别，通常是 READ_COMMITTED 或 REPEATABLE_READ。在某些场景下，我们需要显式地设置事务隔离级别，以满足特定的业务需求。

2. Spring Batch 中的事务管理

Spring Batch 依赖于 PlatformTransactionManager 来管理事务。PlatformTransactionManager 是 Spring 框架中事务管理的核心接口，它定义了事务的创建、提交、回滚等操作。

在 Spring Batch 中，Step 是一个独立的执行单元。每个 Step 都有自己的事务上下文，由 Step 关联的 PlatformTransactionManager 来管理。常见的 PlatformTransactionManager 实现包括：

• DataSourceTransactionManager：用于 JDBC 数据源的事务管理。
• JpaTransactionManager：用于 JPA 持久化的事务管理。
• JmsTransactionManager：用于 JMS 消息队列的事务管理。

Step 的事务属性可以通过 TransactionAttribute 来配置，包括隔离级别、传播行为、超时时间等。

3. ThreadLocal 与事务上下文

ThreadLocal 是一种线程封闭的存储机制，它允许每个线程拥有自己的变量副本。Spring 框架使用 ThreadLocal 来存储事务上下文信息，例如当前事务的连接、状态等。

TransactionSynchronizationManager 是 Spring 框架中用于管理事务同步的工具类。它使用 ThreadLocal 来存储当前线程的事务同步状态，并提供了一系列方法来注册和注销事务同步器。

在传统的线程模型中，每个线程都有自己的栈空间和 ThreadLocal 变量。当一个线程启动一个新的事务时，TransactionSynchronizationManager 会将事务信息存储到当前线程的 ThreadLocal 变量中。当事务结束时，TransactionSynchronizationManager 会清除 ThreadLocal 变量中的事务信息。

4. 虚拟线程（Virtual Threads）的特性

虚拟线程是 Java 21 中引入的一种轻量级线程。与传统的操作系统线程（Platform Threads）相比，虚拟线程的创建和销毁成本非常低，可以在一个操作系统线程上运行大量的虚拟线程。

虚拟线程的关键特性在于其“多路复用”的特性。多个虚拟线程可以共享同一个操作系统线程，当一个虚拟线程阻塞时，Java 运行时环境会将该虚拟线程挂起，并切换到另一个可运行的虚拟线程。这种切换操作非常快速，可以有效地提高 CPU 的利用率。

然而，虚拟线程的特性也带来了一些新的挑战。由于多个虚拟线程共享同一个操作系统线程，因此在某些情况下，ThreadLocal 变量可能会出现问题。

5. 虚拟线程与事务隔离失效

当 Spring Batch 使用虚拟线程作为 StepTaskExecutor 时，可能会出现事务隔离级别传播失效的问题。这是因为虚拟线程共享同一个操作系统线程，如果多个虚拟线程同时访问 TransactionSynchronizationManager 中的 ThreadLocal 变量，可能会导致事务信息被覆盖或篡改。

举例说明：

假设我们有一个 Spring Batch 作业，它包含两个 Step：step1 和 step2。step1 和 step2 都配置了 READ_COMMITTED 隔离级别，并且使用虚拟线程作为 StepTaskExecutor。

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public PlatformTransactionManager transactionManager(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

    @Bean
    public TaskExecutor virtualThreadTaskExecutor() {
        return Executors.newVirtualThreadPerTaskExecutor();
    }

    @Bean
    public Step step1(PlatformTransactionManager transactionManager) {
        return stepBuilderFactory.get("step1")
                .transactionManager(transactionManager)
                .taskExecutor(virtualThreadTaskExecutor())
                .<String, String>chunk(10)
                .reader(new ItemReader<String>() {
                    private int count = 0;
                    @Override
                    public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                        if (count < 100) {
                            count++;
                            return "item" + count;
                        }
                        return null;
                    }
                })
                .processor(item -> {
                    // 模拟一些处理逻辑
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return item.toUpperCase();
                })
                .writer(items -> {
                    // 模拟写入操作，这里可以访问数据库
                    System.out.println("step1 writing: " + items);
                    // 关键：这里可能会受到其他虚拟线程的影响
                    // 假设 step2 也在执行，并且也在访问数据库
                    // 如果 step2 的事务先提交，那么 step1 可能会读取到 step2 提交的数据，导致隔离级别失效
                })
                .build();
    }

    @Bean
    public Step step2(PlatformTransactionManager transactionManager) {
        return stepBuilderFactory.get("step2")
                .transactionManager(transactionManager)
                .taskExecutor(virtualThreadTaskExecutor())
                .<String, String>chunk(10)
                .reader(new ItemReader<String>() {
                    private int count = 0;
                    @Override
                    public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                        if (count < 50) {
                            count++;
                            return "item" + count;
                        }
                        return null;
                    }
                })
                .processor(item -> {
                    // 模拟一些处理逻辑
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return item.toLowerCase();
                })
                .writer(items -> {
                    // 模拟写入操作，这里可以访问数据库
                    System.out.println("step2 writing: " + items);
                    // 关键：这里可能会影响 step1 的事务
                })
                .build();
    }

    @Bean
    public Job job(Step step1, Step step2) {
        return jobBuilderFactory.get("myJob")
                .start(step1)
                .next(step2)
                .build();
    }
}

在这个例子中，step1 和 step2 都在使用虚拟线程，并且都在访问数据库。由于虚拟线程共享同一个操作系统线程，因此 step1 和 step2 可能会同时访问 TransactionSynchronizationManager 中的 ThreadLocal 变量。

如果 step2 的事务先提交，那么 step1 在执行写入操作时，可能会读取到 step2 提交的数据。这会导致 step1 的隔离级别失效，因为它读取到了其他事务尚未提交的数据。

这种现象在并发量较高的情况下更容易发生。

6. 解决方案

解决虚拟线程导致的事务隔离失效问题，可以考虑以下几种方案：

6.1. 使用 Platform Threads

最简单的解决方案是避免使用虚拟线程，改用传统的操作系统线程（Platform Threads）。这样可以保证每个线程都有自己的栈空间和 ThreadLocal 变量，从而避免事务信息被覆盖或篡改。

虽然使用 Platform Threads 可以解决事务隔离问题，但它也会带来一些性能上的损失。Platform Threads 的创建和销毁成本较高，不适合处理大量的并发任务。

6.2. 使用 Scoped Values (Java 20+)

Java 20 引入了 Scoped Values，它提供了一种在线程之间安全地共享数据的机制。与 ThreadLocal 相比，Scoped Values 具有以下优势：

• 不可变性： Scoped Values 是不可变的，一旦设置就不能被修改。这可以避免并发修改导致的问题。
• 安全性： Scoped Values 只能在特定的代码块内访问，超出范围就无法访问。这可以提高数据的安全性。
• 性能： Scoped Values 的性能比 ThreadLocal 更高，因为它不需要进行线程上下文切换。

可以使用 Scoped Values 来存储事务上下文信息，从而避免 ThreadLocal 导致的事务隔离问题。

代码示例 (需要 Java 20+):

// 定义 Scoped Value
private static final ScopedValue<TransactionContext> transactionContext = ScopedValue.newInstance();

// 在事务开始时设置 Scoped Value
public void doInTransaction(Runnable task) {
    TransactionContext context = new TransactionContext(); // 创建事务上下文
    ScopedValue.runWhere(transactionContext, context, task); // 在 Scoped Value 中运行任务
}

// 在任务中获取 Scoped Value
public void myTask() {
    TransactionContext context = transactionContext.get(); // 获取事务上下文
    // 使用事务上下文进行操作
}

// 事务上下文类 (示例)
static class TransactionContext {
    // 存储事务相关信息，例如数据库连接
    private Connection connection;

    public TransactionContext() {
        // 初始化连接
    }

    public Connection getConnection() {
        return connection;
    }
}

6.3. 手动管理事务上下文

另一种解决方案是手动管理事务上下文，避免使用 TransactionSynchronizationManager 自动管理。这种方案需要自己负责事务的创建、提交和回滚，以及事务上下文的传递。

手动管理事务上下文的优点是可以更好地控制事务的生命周期，但缺点是代码复杂度较高，容易出错。

代码示例：

@Component
public class MyItemProcessor implements ItemProcessor<String, String> {

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Override
    public String process(String item) throws Exception {
        TransactionStatus status = null;
        try {
            // 1. 创建事务
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
            status = transactionManager.getTransaction(def);

            // 2. 执行业务逻辑
            String result = item.toUpperCase();
            // 模拟一些数据库操作
            // ...

            // 3. 提交事务
            transactionManager.commit(status);
            return result;
        } catch (Exception e) {
            // 4. 回滚事务
            if (status != null) {
                transactionManager.rollback(status);
            }
            throw e;
        }
    }
}

6.4. 调整 TaskExecutor 的并发度

如果以上方案都不可行，可以尝试调整 TaskExecutor 的并发度。降低并发度可以减少虚拟线程之间的竞争，从而降低事务隔离失效的概率。

可以通过设置 ThreadPoolTaskExecutor 的 corePoolSize 和 maxPoolSize 来调整并发度。

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(1); // 设置核心线程数为 1
    executor.setMaxPoolSize(1);  // 设置最大线程数为 1
    executor.setQueueCapacity(0); // 设置队列容量为 0，避免任务堆积
    executor.initialize();
    return executor;
}

需要注意的是，降低并发度可能会降低作业的整体性能。

6.5. 数据库层面的隔离级别增强

虽然Spring Batch 层面的隔离级别控制可能失效，但是数据库本身的隔离级别依然生效。可以考虑在数据库层面设置更高的隔离级别，例如 SERIALIZABLE，来强制事务串行执行，从而避免并发问题。

但这种方式会显著降低数据库的并发性能，需要谨慎评估。

7. 选择合适的方案

选择哪种方案取决于具体的应用场景和需求。

• 如果对性能要求不高，并且可以接受使用 Platform Threads，那么使用 Platform Threads 是最简单的解决方案。
• 如果使用 Java 20 或更高版本，并且希望获得更高的性能和安全性，那么使用 Scoped Values 是一个不错的选择。
• 如果需要更好地控制事务的生命周期，并且对代码复杂度要求不高，那么手动管理事务上下文也是一个可行的方案。
• 如果以上方案都不可行，可以尝试调整 TaskExecutor 的并发度，或者在数据库层面增强隔离级别。

在实际应用中，可能需要结合多种方案来解决虚拟线程导致的事务隔离失效问题。

8. 代码示例：Scoped Values 的集成

下面是一个更完整的代码示例，展示了如何在 Spring Batch 中集成 Scoped Values 来解决事务隔离问题 (需要 Java 20+):

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private DataSource dataSource;

    // 定义 Scoped Value
    private static final ScopedValue<TransactionContext> transactionContext = ScopedValue.newInstance();

    @Bean
    public PlatformTransactionManager transactionManager() {
        return new DataSourceTransactionManager(dataSource);
    }

    @Bean
    public TaskExecutor virtualThreadTaskExecutor() {
        return Executors.newVirtualThreadPerTaskExecutor();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .taskExecutor(virtualThreadTaskExecutor())
                .<String, String>chunk(10)
                .reader(new MyItemReader())
                .processor(new MyItemProcessor())
                .writer(new MyItemWriter())
                .build();
    }

    @Bean
    public Job job(Step step1) {
        return jobBuilderFactory.get("myJob")
                .start(step1)
                .build();
    }

    // 自定义的 ItemReader
    public class MyItemReader implements ItemReader<String> {
        private int count = 0;

        @Override
        public String read() throws Exception {
            if (count < 100) {
                count++;
                return "item" + count;
            }
            return null;
        }
    }

    // 自定义的 ItemProcessor
    @Component
    public class MyItemProcessor implements ItemProcessor<String, String> {

        @Autowired
        private PlatformTransactionManager transactionManager;

        @Override
        public String process(String item) throws Exception {
            // 1. 创建事务上下文
            TransactionContext context = new TransactionContext(dataSource);

            // 2. 执行事务操作
            return ScopedValue.callWhere(transactionContext, context, () -> {
                TransactionStatus status = null;
                try {
                    // 创建事务
                    DefaultTransactionDefinition def = new DefaultTransactionDefinition();
                    def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
                    status = transactionManager.getTransaction(def);

                    // 执行业务逻辑，并使用事务上下文
                    String result = item.toUpperCase() + " - " + transactionContext.get().getConnection().getMetaData().getDatabaseProductName();
                    // 模拟一些数据库操作
                    // ...

                    // 提交事务
                    transactionManager.commit(status);
                    return result;
                } catch (Exception e) {
                    // 回滚事务
                    if (status != null) {
                        transactionManager.rollback(status);
                    }
                    throw new RuntimeException(e); // 包装异常
                }
            });
        }
    }

    // 自定义的 ItemWriter
    public class MyItemWriter implements ItemWriter<String> {
        @Override
        public void write(List<? extends String> items) throws Exception {
            System.out.println("Writing: " + items);
            // 可以在这里使用 transactionContext.get().getConnection() 进行数据库操作
        }
    }

    // 事务上下文类
    public static class TransactionContext {
        private Connection connection;

        public TransactionContext(DataSource dataSource) {
            try {
                this.connection = DataSourceUtils.getConnection(dataSource);
            } catch (SQLException e) {
                throw new RuntimeException("Failed to get connection", e);
            }
        }

        public Connection getConnection() {
            return connection;
        }
    }
}

在这个示例中，MyItemProcessor 使用 Scoped Value transactionContext 来传递事务上下文信息。在 ScopedValue.callWhere 中，我们创建了一个新的 TransactionContext 对象，并将其设置为当前线程的 Scoped Value。在 process 方法中，我们可以通过 transactionContext.get() 来获取事务上下文对象，并使用其中的数据库连接进行操作。

9. 虚拟线程环境下的Spring Batch

虚拟线程本身是JDK提供的一种新的并发方案，它与Spring Batch集成时，需要考虑虚拟线程的特性，尤其是其对ThreadLocal的影响。如果项目使用的JDK版本较高，可以考虑使用Scoped Values来代替ThreadLocal，从而避免潜在的线程安全问题。同时，需要仔细评估并发度，并根据实际情况选择合适的事务管理策略。

10. 隔离级别传播失效问题的总结

在Spring Batch中使用虚拟线程时，由于虚拟线程的特性，可能会导致事务隔离级别传播失效的问题。为了解决这个问题，我们可以选择使用Platform Threads、Scoped Values、手动管理事务上下文、调整TaskExecutor的并发度，或者在数据库层面增强隔离级别。选择哪种方案取决于具体的应用场景和需求。理解虚拟线程的特性，并结合实际情况选择合适的事务管理策略，是编写健壮的Spring Batch应用的关键。