。

线程池也很少有人会自己去建#xff0c;默认的随便用用。

在工作中大家对…前言

多线程大家肯定都不陌生理论滚瓜烂熟八股天花乱坠但是大家有多少在代码中实践过呢很多人在实际开发中可能就用用Asyncnew

Thread()。

线程池也很少有人会自己去建默认的随便用用。

在工作中大家对于多线程开发大多是用在异步比如发消息但是对于提效这块最重要的优势却很少有人涉及。

因此本篇文章会结合我自己的工作场景带大家去发掘项目中的多线程场景让你的代码快如闪电。

最后会有我对自己三个多月来写的十篇原创文章的个人总结希望读者大大能耐心看完本篇一定会有所收获。

多线程普及

操作系统增加了进程和线程分时复用CPU进而均衡CPU与IO设备的速度差异–导致了原子性问题

可见性问题–线程A修改共享变量修改后CPU缓存中的数据没有及时同步到内存线程B读取了内存中老数据

有序性问题–经典的对象创建三步堆中分配内存–初始化–变量指向内存地址如果排序重排会出现132导致没有初始化的对象被创建

JVM提供了什么工具去解决线程不安全问题Java代码有哪些实现思路

JVM提供了三个关键词synchronized、volatile、final和JMM(线程操作内存规范例如8个happen

before原则)

同步synchronized和ReentrantLock非同步CAS(CPU原语依赖硬件)线程安全局部变量(虚拟机栈或者本地方法栈线程私有)和ThreadLocal(本地线程变量副本空间换安全每个线程一份)

如何开启线程

基础的Thread、runable、callable进阶的ThreadExecutor和Future以及JDK8的终极武器CompletableFuture

线程间如何协作

volatile关键字用来修饰共享变量保证了共享变量的可见性任何线程需要读取时都要到内存中读取确保获得最新值。

synchronized关键字确保只能同时有一个线程访问方法或者变量保证了线程访问的可见性和排他性。

等待/通知机制–指一个线程A调用了对象O的wait()方法进入等待状态而另一个线程B

调用了对象O的notify()或者notifyAll()方法线程A收到通知后从对象O的wait()方法返回进而

执行后续操作管道输入/输出流

管道输入/输出流和普通的文件输入/输出流或者网络输入/输出流不同之处在于它主要

用于线程之间的数据传输而传输的媒介为内存。

管道输入/输出流主要包括了如下4种具体实现PipedOutputStream、PipedInputStream、

PipedReader和PipedWriter前两种面向字节而后两种面向字符。

join()方法–如果一个线程A执行了thread.join()语句其含义是当前线程A等待thread线程终止之后才

从thread.join()返回。

线程Thread除了提供join()方法之外还提供了join(long

millis)和join(long

nanos)两个具备超时特性的方法。

这两个超时方法表示如果线程thread在给定的超时

时间里没有终止那么将会从该超时方法中返回。

ThreadLocal–即线程本地变量每个线程都有自己唯一的一个哦是一个以ThreadLocal对象为键、任意对象为值的存储结构。

底层是一个ThreadLocalMap来存储信息key是弱引用value是强引用所以使用完毕后要及时清理(尤其使用线程池时)。

有JDK5开始提供的Semaphore(信号量)、CyclicBarrier、CountDownLatch以及JDK8的CompletableFuture

场景实战

以下案例主要运用CompletableFuture.allOf()方法将原本串行的操作改为并行。

本案例相对比较常规算是CompletableFuture的基本操作其他特性就不一一介绍了。

AtomicReferenceListSellOrderList

orderLists

AtomicReferenceListInternationalSalesList

salesLists

{orderLists.set(sellOrderListService.lambdaQuery().ge(SellOrderList::getOrderCreateDate,

startDate).le(SellOrderList::getOrderCreateDate,

endDate).eq(SellOrderList::getIsDelete,

0).list());});

CompletableFuture.allOf(orderListCom,

productCom,

{//获取全部数据后处理数据aggregateData(customerList,

saleMonth,

这里借鉴了parallelStream流的思路将串行的for循环分割成多个集合后对分割后的集合进行循环。

这应该是最普遍的多线程应用场景了需要注意的是线程池需要自定义大小、不安全的集合例如ArrayList并行add时需要加锁加好日志就完事了。

Autowired

//自建线程池ForkJoinPool默认的太小一般是逻辑CPU数量-1

int

Runtime.getRuntime().availableProcessors();

ForkJoinPool

com.google.common.collect.Lists;

ListListSlErpDTO

CompletableFuture.allOf(partition.stream().map(addPartitionList

{for

{//ArrayList线程不安全多线程会出现数据覆盖体现为数据丢失slAddList.add(sl);}}},

ioDense)).toArray(CompletableFuture[]::new)).whenComplete((res,

{if

{//进一步处理循环后的结果slService.batchSchedule(versionNum,

slAddList);}

一个常见的场景当我们需要判断当前数据的唯一值是否在数据库存在从而判断是新增还是修改时常规做法是

for(X

xKeyx.getKey();//key为当前数据的唯一值for(Y

y:数据库中的数据){if(Objects.equals(xKey,yKey)){update;//相等就修改

}else{add;//不等就新增

循环套循环时间复杂度O(n²)性能肯定是比较差的那怎么改造呢如果能确定key是唯一值就拿优化神器Map来操作算法里面常见的优化手段。

我们把唯一值key作为键输出对象作为值那么上面的代码就可以改造为

public

lambdaQuery().eq(SafeRule::getIsDelete,

0).list();return

ruleList.stream().collect(Collectors.toMap(e

e.getBigClass()

xKeyx.getKey();//key为当前数据的唯一值MapString,

SafeRule

maplist2Map();if(map.get(xKey)!null){update;//相等就修改

}else{add;//不等就新增

引入Map之后第二个嵌套的循环就干掉了换成了单次循环来组装Map时间复杂度降为O(n)。

此方法仅适用于每行数据拥有唯一值不然stream在组装时会提示重复key。

修改Map遍历为并行操作

ArrayList();list.add(splitMap);return

list;}Setk

(!CollectionUtils.isEmpty(tem))

total;

materialList.parallelStream().collect(Collectors.groupingBy(e

e.getInvOrgId()

e.getVendorId()));ListMapString,

ListBudgetErpDTO

CompletableFuture.allOf(mapList.stream().map(splitMap

list)

ioDense)).toArray(CompletableFuture[]::new)).exceptionally(e

e);return

null;});上面提供了一个切分的工具类以及Map改造的代码总体还是非常简单思路和for循环的改造是差不多的

多线程新增

我个人在开发中会使用一些小工具来提高开发效率接下来公开一个我常用的批量插入的小工具这个小工具最开始是同事给我的然后我做了优化和扩充主要是扩充了多线程以及service块的代码。

该工具类用于生成复制可用的代码这里需要提前指定一些固定变量。

除了entity和serviceName需要根据实际情况变化之外方法名和参数名可以不变。

生成了四个方法分别是mapper类中的方法、mapper.xml中的foreach批量插入代码、普通无事务的多线程批量插入代码、多线程事务代码

String

printMapper(entity.getSimpleName(),

methodName,

printSave(entity.getSimpleName(),

serviceName,

printAddTransaction(entity.getSimpleName(),

paramName,

这个多线程插入其实就是我上面多线程处理场景中for循环改造的变种将集合拆分进行并行批量插入

(CollectionUtils.isNotEmpty(addList))

{ListListBudgetBase

1000);CompletableFuture.allOf(partition.stream().map(addPartitionList

baseInfoService.getBaseMapper().batchSchedule(addPartitionList))).toArray(CompletableFuture[]::new)).exceptionally(e

{log.error(多线程处理异常,

这个才是本文的重点接下来我会详细介绍我在开发中遇到的坑和知识点敲黑板了啊重点来了

我写的这个多线程事务本质就是根据2PC理论手写了一个分布式事务涉及到多线程、Spring事务、ThreadLocal、LockSupport这些知识点在线上一定要慎重使用最好不用可作炫技用秀就完了。

从头说起既然是多线程事务那自然不能使用注解Transactional去开启事务Spring事务采用ThreadLocal来做线程隔离ThreadLocalMap内部key为当前线程的ThreadLocal对象也可以当作以当前线程为keyvalue也是个map看源码可以知道map里面key为数据源value为数据库连接。

当然上来看源码肯定认识不够深刻接下来是一段错误代码示范充分展示了理解上面那段话的重要性。

我的第一次失败就是如下一段代码首先肯定是能运行的不能运行的例子我就不给了先来看看这段代码。

//存储事务集合

DefaultTransactionDefinition();

def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

def.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);

TransactionStatus

transactionManager.getTransaction(def);

traStatusList.add(statusStart);

//外部DML操作

lambdaUpdate().set(RedundantErpSl::getIsDelete,

1).set(RedundantErpSl::getUpdateTime,

new

Date()).eq(RedundantErpSl::getVersionNum,

versionNumber).eq(RedundantErpSl::getIsDelete,

0).update();

Lists.partition(RedundantErpSlList,

1000);

CompletableFuture.allOf(partition.stream().map(addPartitionList

{//Spring事务内部由ThreadLocal存储事务绑定信息因此需要每个线程新开一个事务DefaultTransactionDefinition

defGo

DefaultTransactionDefinition();defGo.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);defGo.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);TransactionStatus

statusGo

transactionManager.getTransaction(defGo);//ArrayList线程不安全多线程会出现数据覆盖体现为数据丢失synchronized

(traStatusList)

{traStatusList.add(statusGo);}getBaseMapper().batchSchedule(addPartitionList);})).toArray(CompletableFuture[]::new)).exceptionally(e

{log.error(批量导入出现异常,

PtmException(e.getMessage());});future.join();for

(TransactionStatus

{transactionManager.commit(status);}

catch

{transactionManager.rollback(status);}

}先说说这个错误例子我当时开发的思路手动开启事务后在每个线程操作开始的时候都创建一个事务Spring事务传播级别用的TransactionDefinition.PROPAGATION_REQUIRES_NEW即默认创建新事务。

隔离级别一开始没改然后我就尝试着操作了一下好家伙批量新增的时候直接锁了。

SELECT

INFORMATION_SCHEMA.INNODB_LOCKS;

SELECT

INFORMATION_SCHEMA.INNODB_LOCK_WAITS;

异常如下图锁超时异常

第一次看见下图这个错的时候我是疑惑的没有当回事以为是多数据源的问题。

我项目里有直连Oracle和MySQL两种关系型数据库当时怀疑是多数据源事务没有正确解绑导致的问题。

:::info

PS事实上这个坑给足了我提示根本原因就是多线程事务解绑失败但是我理解出现了偏差为后文埋下了伏笔。

:::

我当时一看有锁这我能惯着马上修改事务隔离级别为TransactionDefinition.ISOLATION_READ_COMMITTED读已提交(MySQL默认事务隔离级别为可重复读这里下调了一级总共四级)。

顺利插入但还是报上面这个错错误位置是下面这个循环提交时报的第二次循环的时候一定会报错。

for

{transactionManager.commit(status);

}当时一度以为是多数据源的问题但是Debug后发现resource里面只有一个数据源key解绑一次后就没了第二个循环解绑的时候就报上面这个错因为找不到可以解绑的key了。

我就很疑惑为啥就一个数据源key我不是在别的线程开了事务嘛按理说开了多少个线程就有多少个事务这个问题困扰了我大概一天左右的时间。

然后我想到了Spring事务的实现原理ThreadLocal然后联想到我的多线程开启事务再看到我在主线程里面进行傻叉循环解绑我瞬间为梦想窒息。

所以破案了我在主线程是操作不了子线程事务这也是代码报key找不到的原因因为用主线程做key在ThreadLocal里肯定是拿不到子线程信息的只能拿到主线程自己的。

多线程事务提交方案

因此解决方案就很简单子线程的事务自己操作那么多线程事务处理哪家强JDK里找CompletableFuture当然这里使用CountDownLatch也是可行的网上也有案例。

多线程事务在处理逻辑上其实和分布式事务很像因此我这里采用2PC的思想一阶段所有子线程全部开启事务并执行SQL然后阻塞等待二阶段判断是否全部成功是就唤醒所有线程提交事务否就全部回滚。

Autowired

-----------多线程事务新增操作-----------

private

(!CollectionUtils.isEmpty(addList))

{//定义局部变量是否成功、顺序标识、等待线程队列AtomicBoolean

isSuccess

AtomicBoolean(true);AtomicInteger

cur

ArrayList();//切分新增集合ListListBudgetBase

partition

partition.size();//多线程处理开始CompletableFutureVoid

future

CompletableFuture.allOf(partition.stream().map(addPartitionList

{//Spring事务内部由ThreadLocal存储事务绑定信息因此需要每个线程新开一个事务DefaultTransactionDefinition

defGo

DefaultTransactionDefinition();defGo.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);TransactionStatus

statusGo

transactionManager.getTransaction(defGo);int

curInt

Thread.currentThread().getName());baseInfoService.getBaseMapper().batchSchedule(addPartitionList);log.info(当前是第{}个线程完成批量插入开始加入等待队列线程名{},

curInt,

Thread.currentThread().getName());//ArrayList线程不安全多线程会出现数据覆盖体现为数据丢失synchronized

(unfinishedList)

{unfinishedList.add(Thread.currentThread());}log.info(当前是第{}个线程已加入队列开始休眠线程名{},

curInt,

Thread.currentThread().getName());notifyAllThread(unfinishedList,

totalSize,

Thread.currentThread().getName());transactionManager.commit(statusGo);}

else

Thread.currentThread().getName());transactionManager.rollback(statusGo);}}

catch

Thread.currentThread().getName(),

e);transactionManager.rollback(statusGo);isSuccess.set(false);notifyAllThread(unfinishedList,

totalSize,

ioDense)).toArray(CompletableFuture[]::new));future.join();}}

private

{log.info(唤醒当前所有休眠线程线程数{}总线程数{},是否强制{},

totalSize,

thread.getName());LockSupport.unpark(thread);}}}方案详解

为什么用LockSupport的park()和unpark()而不用Thread.sleep()、Object.wait()、Condition.await()?

更简单不需要获取锁能直接阻塞线程。

更直观以thread为操作对象更符合阻塞线程的直观定义更精确可以准确地唤醒某一个线程notify随机唤醒一个线程notifyAll唤醒所有等待的线程更灵活

第4点很重要如果不能提前使用unpark()的话按照代码逻辑最后一个线程会被永久阻塞。

为什么要自建线程池

CompletableFuture默认的线程池ForkJoinPool.commonPool()偏向于计算密集型任务处理核心线程数和逻辑CPU数少1对于多线程事务这种IO密集型任务来说核心线程数偏少。

并且上述方法在操作中都是阻塞线程无法一次性开启全部线程的话会导致notifyAllThread方法无法执行老线程阻塞新线程无法开启就尬住了。

ForkJoinPool基于工作窃取算法所以最适合的是计算密集型任务这里我们开启一个参数调整为IO密集型(多核心少队列)的ThreadPoolTaskExecutor线程池即可。

使用多线程的时候也别忘了调整其他组件的最大连接数。

Druid线程池这个代码配置可以调MySQL5.7默认151得用配置文件调整。

MySQL最大连接数调整的方法之前从零开始的SQL修炼手册-实战篇有讲解过欢迎读者们翻翻我之前写的干货。

Mybatis与JDBC批量插入MySQL数据库性能测试及解决方案

先贴一个大佬的文章里面大概讲了Mybatis批量插入和JDBC批量手动事务的优劣结论就是小于1W用Mybatis小于10W且大于1W用JDBC大于10W必须数据分批。

当然我上面多线程做的分批导入肯定是不安全的操作如果要一次性导入的话根据数据量判断用大佬文章中的代码案例即可。

多线程事务提交方案–使用须知

如果此时数据库连接池配置较小比如spring.datasource.druid.max-active10(Druid配置最大连接数为10)。

但是我在使用多线程提交时分批次数为20那么开了10个之后达到上线就会一直卡住原因是老的线程挂起不会释放新的线程因为线程池满了无法创建。

因此在使用该方案时一定要估算数据量分好合适的大小连接池和数据库的最大连接数也要注意是否匹配。