使用parallelStream时数据源错误

[hunao0221]

service本身用了@DS(DsConstant.BASE_TENANT)只操作租户数据库，在使用parallelStream遍历时发现遍历部分数据后，突然变成操作lamp_ds_c_defaults库了

 // 伪代码
 final Long tenantId = ContextUtil.getTenantId();
 items.parallelStream().forEach(i -> {
      
     // ContextUtil.setTenantId(tenantId);
     // 如果不重新设置tenantId，大概率会操作默认数据库 lamp_ds_c_defaults
     service.saveToTenantDB(i);
)}

如日志所示，前两次还是正常的，第三次就出问题了

[lamp-base-server:18764::] 2023-09-14 12:48:56.241:[ INFO] [39213] [cd4fee5855fd46fe8358ebcbbcb1ef9c] [XNIO-2 task-1:80772] [t.t.l.base.handler.customer.CustomerImportHandler.lambda$process$0:90] 单行耗时: 15 毫秒
[lamp-base-server:18764::] 2023-09-14 12:48:56.248:[ INFO] [39213] [cd4fee5855fd46fe8358ebcbbcb1ef9c] [XNIO-2 task-1:80779] [t.t.l.base.handler.customer.CustomerImportHandler.lambda$process$0:90] 单行耗时: 7 毫秒
[lamp-base-server:18764::] 2023-09-14 12:48:56.303:[ WARN] [39213] [cd4fee5855fd46fe8358ebcbbcb1ef9c] [XNIO-2 task-1:80834] [t.t.b.boot.handler.AbstractGlobalExceptionHandler.otherExceptionHandler:245] Exception:

org.springframework.jdbc.BadSqlGrammarException: 
### Error querying database.  Cause: java.sql.SQLSyntaxErrorException: Table 'lamp_ds_c_defaults.dm_customer_org' doesn't exist
### The error may exist in top/tangyh/lamp/base/mapper/customer/CustomerOrgMapper.java (best guess)
### The error may involve defaultParameterMap
### The error occurred while setting parameters
### SQL: SELECT  id,name_,level_,tree_grade,tree_path,address,building_type,room_count,parent_id,sort_value,updated_time,updated_by,created_time,created_by  FROM dm_customer_org      WHERE  (level_ = ? AND parent_id = ? AND name_ = ?)
### Cause: java.sql.SQLSyntaxErrorException: Table 'lamp_ds_c_defaults.dm_customer_org' doesn't exist
; bad SQL grammar []; nested exception is java.sql.SQLSyntaxErrorException: Table 'lamp_ds_c_defaults.dm_customer_org' doesn't exist
	at org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator.doTranslate(SQLErrorCodeSQLExceptionTranslator.java:239) ~[spring-jdbc-5.3.29.jar:5.3.29]
	at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:70) ~[spring-jdbc-5.3.29.jar:5.3.29]
	at org.mybatis.spring.MyBatisExceptionTranslator.translateExceptionIfPossible(MyBatisExceptionTranslator.java:92) ~[mybatis-spring-2.1.1.jar:2.1.1]
	at org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:439) ~[mybatis-spring-2.1.1.jar:2.1.1]
	at com.sun.proxy.$Proxy184.selectList(Unknown Source) ~[na:na]
	at org.mybatis.spring.SqlSessionTemplate.selectList(SqlSessionTemplate.java:224) ~[mybatis-spring-2.1.1.jar:2.1.1]
	at com.baomidou.mybatisplus.core.override.MybatisMapperMethod.executeForMany(MybatisMapperMethod.java:166) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at com.baomidou.mybatisplus.core.override.MybatisMapperMethod.execute(MybatisMapperMethod.java:77) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at com.baomidou.mybatisplus.core.override.MybatisMapperProxy$PlainMethodInvoker.invoke(MybatisMapperProxy.java:148) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at com.baomidou.mybatisplus.core.override.MybatisMapperProxy.invoke(MybatisMapperProxy.java:89) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at com.sun.proxy.$Proxy245.selectList(Unknown Source) ~[na:na]
	at com.baomidou.mybatisplus.core.mapper.BaseMapper.selectOne(BaseMapper.java:191) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at java.base/java.lang.invoke.MethodHandle.invokeWithArguments(MethodHandle.java:710) ~[na:na]
	at com.baomidou.mybatisplus.core.override.MybatisMapperProxy$DefaultMethodInvoker.invoke(MybatisMapperProxy.java:162) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at com.baomidou.mybatisplus.core.override.MybatisMapperProxy.invoke(MybatisMapperProxy.java:89) ~[mybatis-plus-core-3.5.3.2.jar:3.5.3.2]
	at com.sun.proxy.$Proxy245.selectOne(Unknown Source) ~[na:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
	at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[na:na]
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:137) ~[spring-tx-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:241) ~[spring-aop-5.3.29.jar:5.3.29]
	at com.sun.proxy.$Proxy246.selectOne(Unknown Source) ~[na:na]
	at com.baomidou.mybatisplus.extension.service.impl.ServiceImpl.getOne(ServiceImpl.java:209) ~[mybatis-plus-extension-3.5.3.2.jar:3.5.3.2]
	at com.baomidou.mybatisplus.extension.service.IService.getOne(IService.java:327) ~[mybatis-plus-extension-3.5.3.2.jar:3.5.3.2]
	at com.baomidou.mybatisplus.extension.service.IService$$FastClassBySpringCGLIB$$f8525d18.invoke(<generated>) ~[mybatis-plus-extension-3.5.3.2.jar:3.5.3.2]
	at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy.invokeMethod(CglibAopProxy.java:386) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy.access$000(CglibAopProxy.java:85) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:704) ~[spring-aop-5.3.29.jar:5.3.29]
	at top.tangyh.lamp.base.manager.customer.impl.CustomerOrgManagerImpl$$EnhancerBySpringCGLIB$$69afe7b5.getOne(<generated>) ~[classes/:na]
	at top.tangyh.lamp.base.service.customer.impl.CustomerOrgServiceImpl.findByNameLevel(CustomerOrgServiceImpl.java:118) ~[classes/:na]
	at top.tangyh.lamp.base.service.customer.impl.CustomerOrgServiceImpl$$FastClassBySpringCGLIB$$630e67cd.invoke(<generated>) ~[classes/:na]
	at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:123) ~[spring-tx-5.3.29.jar:5.3.29]
	at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:388) ~[spring-tx-5.3.29.jar:5.3.29]
	at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) ~[spring-tx-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.29.jar:5.3.29]
	at com.baomidou.dynamic.datasource.aop.DynamicDataSourceAnnotationInterceptor.invoke(DynamicDataSourceAnnotationInterceptor.java:56) ~[dynamic-datasource-spring-4.1.3.jar:na]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.29.jar:5.3.29]
	at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708) ~[spring-aop-5.3.29.jar:5.3.29]
	at top.tangyh.lamp.base.service.customer.impl.CustomerOrgServiceImpl$$EnhancerBySpringCGLIB$$f1a25b32.findByNameLevel(<generated>) ~[classes/:na]
	at top.tangyh.lamp.base.handler.customer.CustomerImportHandler.getOrgId(CustomerImportHandler.java:215) ~[classes/:na]
	at top.tangyh.lamp.base.handler.customer.CustomerImportHandler.lambda$processAreaOrg$5(CustomerImportHandler.java:200) ~[classes/:na]
	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1705) ~[na:na]
	at top.tangyh.lamp.base.handler.customer.CustomerImportHandler.processAreaOrg(CustomerImportHandler.java:199) ~[classes/:na]
	at top.tangyh.lamp.base.handler.customer.CustomerImportHandler.lambda$process$0(CustomerImportHandler.java:78) ~[classes/:na]
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[na:na]
	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1655) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484) ~[na:na]
	at java.base/java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:290) ~[na:na]
	at java.base/java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:746) ~[na:na]
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290) ~[na:na]
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020) ~[na:na]
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656) ~[na:na]
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594) ~[na:na]
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183) ~[na:na]

[zuihou]

不要在for里面操作数据库，尤其是parallelStream。

改成单线程操作数据库，或者先for处理逻辑，最后批量操作数据库。