Spring boot，多原则，多模块，@Transactional，parallelStream答案

【问题标题】：Spring boot, multi-tenet, multi-module, @Transactional , parallelStreamSpring boot，多原则，多模块，@Transactional，parallelStream
【发布时间】：2020-04-13 10:19:01
【问题描述】：

我正在尝试将一些 50k 记录插入到数据库中。我们使用了 AbstractRoutingDataSource，它使用 TenantContext 解析数据源，TenantContext 是一个实用程序类，具有私有静态 final ThreadLocal CURRENT_TENANT = new ThreadLocal();

当我使用并行流或尝试使用 @Async 方法时，我收到以下错误

代码：

             .parallelStream()
             .forEach(row -> {
                 TenantContext.setCurrentTenant(centerCd);
                 someDao.insert(row);
             });

错误：


org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.lang.IllegalStateException: Cannot determine target DataSource for lookup key [null]
    at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:305)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:378)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.createTransactionIfNecessary(TransactionAspectSupport.java:474)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:289)
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:98)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.aop.interceptor.AsyncExecutionInterceptor.lambda$invoke$0(AsyncExecutionInterceptor.java:115)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException: Cannot determine target DataSource for lookup key [null]
    at org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource.determineTargetDataSource(AbstractRoutingDataSource.java:207)
    at org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource.getConnection(AbstractRoutingDataSource.java:169)
    at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:262)
    ... 10 common frames omitted

【问题讨论】：

标签： spring multithreading spring-boot transactions multi-module

【解决方案1】：

它的工作方式与您描述的完全一样：您的 TenantContext 正是 ThreadLocal 并且存在于一个线程中，该线程由 parallelStream() 或 Async 方法启动。（实际上，Async 或forEach 方法内部的调用是来自Runnable 的run）

在线程开始时尝试注入/解析数据源：因为您的事务必须在创建线程时开始，在您的Runnable 进入run 方法之前。而此时您还没有指定您的租户，调用TenantContext.setCurrentTenant(centerCd) 稍后在run 方法实现中执行。

我建议将这种结构应用于您的代码：

class TenantAwareThread extends Thread {

    public TenantAwareThread(Runnable target, TenantData tenantData) {
        super(target);
        TenantContext.setCurrentTenant(tenantData);
    }
}

@Autowired
TaskExecutor executor;

void startTask(TenantData tenantData, RowData row) {
    executor.execute(
        new TenantAwareThread(() -> {
            someDao.insert(row);
        },
        tenantData));
}

您创建一个从一开始就知道租户数据的新线程类型。并简单地将您的执行包装到这样的线程中。

【讨论】：