北京网站制作公司哪家好,开发手机网站制作,wordpress返利,网站开发费走什么科目前言
多线程大家肯定都不陌生#xff0c;理论滚瓜烂熟#xff0c;八股天花乱坠#xff0c;但是大家有多少在代码中实践过呢#xff1f;很多人在实际开发中可能就用用Async#xff0c;new Thread()。线程池也很少有人会自己去建#xff0c;默认的随便用用。在工作中大家对…前言
多线程大家肯定都不陌生理论滚瓜烂熟八股天花乱坠但是大家有多少在代码中实践过呢很多人在实际开发中可能就用用Asyncnew Thread()。线程池也很少有人会自己去建默认的随便用用。在工作中大家对于多线程开发大多是用在异步比如发消息但是对于提效这块最重要的优势却很少有人涉及。因此本篇文章会结合我自己的工作场景带大家去发掘项目中的多线程场景让你的代码快如闪电。 最后会有我对自己三个多月来写的十篇原创文章的个人总结希望读者大大能耐心看完本篇一定会有所收获。
多线程普及
多线程解决了什么问题带来了什么问题
Cpu为了均衡与内存的速度差异增加了缓存–导致了可见性问题 操作系统增加了进程和线程分时复用CPU进而均衡CPU与IO设备的速度差异–导致了原子性问题 编译程序优化指令排序(JVM指令重排序)–导致了有序性问题
可见性问题–线程A修改共享变量修改后CPU缓存中的数据没有及时同步到内存线程B读取了内存中老数据 原子性问题–多个线程增加数据有几个线程挂了这数据就少了 有序性问题–经典的对象创建三步堆中分配内存–初始化–变量指向内存地址如果排序重排会出现132导致没有初始化的对象被创建
JVM提供了什么工具去解决线程不安全问题Java代码有哪些实现思路
JVM提供了三个关键词synchronized、volatile、final和JMM(线程操作内存规范例如8个happen before原则) Java代码实践可从三方面入手
同步synchronized和ReentrantLock非同步CAS(CPU原语依赖硬件)线程安全局部变量(虚拟机栈或者本地方法栈线程私有)和ThreadLocal(本地线程变量副本空间换安全每个线程一份)
如何开启线程
基础的Thread、runable、callable进阶的ThreadExecutor和Future以及JDK8的终极武器CompletableFuture
线程间如何协作
基础
volatile和synchronized关键字 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 millis,int nanos)两个具备超时特性的方法。这两个超时方法表示如果线程thread在给定的超时 时间里没有终止那么将会从该超时方法中返回。ThreadLocal–即线程本地变量每个线程都有自己唯一的一个哦是一个以ThreadLocal对象为键、任意对象为值的存储结构。底层是一个ThreadLocalMap来存储信息key是弱引用value是强引用所以使用完毕后要及时清理(尤其使用线程池时)。
进阶 有JDK5开始提供的Semaphore(信号量)、CyclicBarrier、CountDownLatch以及JDK8的CompletableFuture
场景实战
多线程处理场景
并行聚合处理数据
以下案例主要运用CompletableFuture.allOf()方法将原本串行的操作改为并行。本案例相对比较常规算是CompletableFuture的基本操作其他特性就不一一介绍了。
AtomicReferenceListSellOrderList orderLists new AtomicReference();
AtomicReferenceListInternationalSalesList salesLists new AtomicReference();
AtomicReferenceMapString, BigDecimal productMap new AtomicReference();
.........
//逻辑A
CompletableFutureVoid orderListCom CompletableFuture.runAsync(() - {orderLists.set(sellOrderListService.lambdaQuery().ge(SellOrderList::getOrderCreateDate, startDate).le(SellOrderList::getOrderCreateDate, endDate).eq(SellOrderList::getIsDelete, 0).list());});
CompletableFutureVoid productCom CompletableFuture.runAsync(() - {//逻辑B});
CompletableFutureVoid euLineCom CompletableFuture.runAsync(() - {//逻辑C});
//汇总线程操作
CompletableFuture.allOf(orderListCom, productCom, euCloudCom).handle((res, e) - {if (e ! null) {log.error(客户订单定时任务聚合数据异常, e);} else {try {//获取全部数据后处理数据aggregateData(customerList, saleMonth, orderLists, salesLists, productMap, euLineList, asLineList,euCloudList, asCloudList, itemMap, deliveryMap, parities);} catch (Exception ex) {log.error(客户订单处理数据异常, ex);}}return null;
});修改for循环为并行操作
这里借鉴了parallelStream流的思路将串行的for循环分割成多个集合后对分割后的集合进行循环。这应该是最普遍的多线程应用场景了需要注意的是线程池需要自定义大小、不安全的集合例如ArrayList并行add时需要加锁加好日志就完事了。
Autowired
Qualifier(ioDenseExecutor)
private ThreadPoolTaskExecutor ioDense;
//自建线程池ForkJoinPool默认的太小一般是逻辑CPU数量-1
int logicCpus Runtime.getRuntime().availableProcessors();
ForkJoinPool forkJoinPool new ForkJoinPool(logicCpus * 80);
//指定集合大小避免频繁扩容
ListRedundantErpSl slAddList new ArrayList(50000);
//谷歌提供工具类切分集合--import com.google.common.collect.Lists;
ListListSlErpDTO partition Lists.partition(slErpList, 1000);
int finalLastStatus lastStatus;
CompletableFutureVoid handle CompletableFuture.allOf(partition.stream().map(addPartitionList -CompletableFuture.runAsync(() - {for (SlErpDTO slErp : addPartitionList) {//TODO 逻辑处理synchronized (slAddList) {//ArrayList线程不安全多线程会出现数据覆盖体现为数据丢失slAddList.add(sl);}}}, ioDense)).toArray(CompletableFuture[]::new)).whenComplete((res, e) - {if (e ! null) {log.error(多线程组装数据失败, e);} else {try {//进一步处理循环后的结果slService.batchSchedule(versionNum, slAddList);} catch (Exception ex) {log.error(批量插入失败, ex);}}});
handle.join();List转Map解除嵌套循环
一个常见的场景当我们需要判断当前数据的唯一值是否在数据库存在从而判断是新增还是修改时常规做法是
for(X x:待新增或者修改的数据){String xKeyx.getKey();//key为当前数据的唯一值for(Y y:数据库中的数据){if(Objects.equals(xKey,yKey)){update;//相等就修改 }else{add;//不等就新增 }}
} 循环套循环时间复杂度O(n²)性能肯定是比较差的那怎么改造呢如果能确定key是唯一值就拿优化神器Map来操作算法里面常见的优化手段。我们把唯一值key作为键输出对象作为值那么上面的代码就可以改造为
public MapString, SafeRule list2Map() {ListSafeRule ruleList lambdaQuery().eq(SafeRule::getIsDelete, 0).list();return ruleList.stream().collect(Collectors.toMap(e - e.getBigClass() e.getSmallClass(),
Function.identity()));
}
for(X x:待新增或者修改的数据){String xKeyx.getKey();//key为当前数据的唯一值MapString, SafeRule maplist2Map();if(map.get(xKey)!null){update;//相等就修改 }else{add;//不等就新增 }}
} 引入Map之后第二个嵌套的循环就干掉了换成了单次循环来组装Map时间复杂度降为O(n)。此方法仅适用于每行数据拥有唯一值不然stream在组装时会提示重复key。 修改Map遍历为并行操作 既然for循环能转换那么map遍历必然也能通过多线程改造。
/*** 将map切成段--工具类** param splitMap 被切段的map* param splitNum 每段的大小*/
public static k, v ListMapk, v mapSplit(Mapk, v splitMap, int splitNum) {if (splitMap null || splitNum 0) {ListMapk, v list new ArrayList();list.add(splitMap);return list;}Setk keySet splitMap.keySet();Iteratork iterator keySet.iterator();int i 1;ListMapk, v total new ArrayList();Mapk, v tem new HashMap();while (iterator.hasNext()) {k next iterator.next();tem.put(next, splitMap.get(next));if (i splitNum) {total.add(tem);tem new HashMap();i 0;}i;}if (!CollectionUtils.isEmpty(tem)) {total.add(tem);}return total;
}
//代码示例
MapString, ListBudgetErpDTO materialMap materialList.parallelStream().collect(Collectors.groupingBy(e - e.getInvOrgId() - e.getItemId() - e.getVendorId()));ListMapString, ListBudgetErpDTO mapList MapUtil.mapSplit(materialMap, 50);CompletableFutureVoid handle CompletableFuture.allOf(mapList.stream().map(splitMap - CompletableFuture.runAsync(() - {splitMap.forEach((identity, list) - {//业务操作});}, ioDense)).toArray(CompletableFuture[]::new)).exceptionally(e - {log.error(多线程组装数据失败, e);return null;});上面提供了一个切分的工具类以及Map改造的代码总体还是非常简单思路和for循环的改造是差不多的
多线程新增
我个人在开发中会使用一些小工具来提高开发效率接下来公开一个我常用的批量插入的小工具这个小工具最开始是同事给我的然后我做了优化和扩充主要是扩充了多线程以及service块的代码。
总览
该工具类用于生成复制可用的代码这里需要提前指定一些固定变量。除了entity和serviceName需要根据实际情况变化之外方法名和参数名可以不变。生成了四个方法分别是mapper类中的方法、mapper.xml中的foreach批量插入代码、普通无事务的多线程批量插入代码、多线程事务代码
//批量方法名对应mapper和xml中id
String methodName batchSchedule;
//mapper参数名称
String paramName addList;
//实际代码里面的service命名
String serviceName baseInfoService;
Class? entity BudgetBase.class;
//批量插入
printMapper(entity.getSimpleName(), methodName, paramName);
printXml(entity, methodName, paramName);
//普通多线程批量插入无事务
printSave(entity.getSimpleName(), serviceName, paramName, 1000);
//多线程事务慎用
printAddTransaction(entity.getSimpleName(), paramName, 1000);mapper方法 xml批量插入语句 多线程批量插入
这个多线程插入其实就是我上面多线程处理场景中for循环改造的变种将集合拆分进行并行批量插入
if (CollectionUtils.isNotEmpty(addList)) {ListListBudgetBase partition Lists.partition(addList, 1000);CompletableFuture.allOf(partition.stream().map(addPartitionList -CompletableFuture.runAsync(() - baseInfoService.getBaseMapper().batchSchedule(addPartitionList))).toArray(CompletableFuture[]::new)).exceptionally(e - {log.error(多线程处理异常, e);return null;});}花里胡哨-多线程事务提交
这个才是本文的重点接下来我会详细介绍我在开发中遇到的坑和知识点敲黑板了啊重点来了 我写的这个多线程事务本质就是根据2PC理论手写了一个分布式事务涉及到多线程、Spring事务、ThreadLocal、LockSupport这些知识点在线上一定要慎重使用最好不用可作炫技用秀就完了。
深刻理解Spring事务、ThreadLocal
从头说起既然是多线程事务那自然不能使用注解Transactional去开启事务Spring事务采用ThreadLocal来做线程隔离ThreadLocalMap内部key为当前线程的ThreadLocal对象也可以当作以当前线程为keyvalue也是个map看源码可以知道map里面key为数据源value为数据库连接。 当然上来看源码肯定认识不够深刻接下来是一段错误代码示范充分展示了理解上面那段话的重要性。我的第一次失败就是如下一段代码首先肯定是能运行的不能运行的例子我就不给了先来看看这段代码。
//存储事务集合
ListTransactionStatus traStatusList new ArrayList();
//最外部更新或者删除时手动创建一个新事务
DefaultTransactionDefinition def new DefaultTransactionDefinition();
def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
def.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
TransactionStatus statusStart 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();
ListListRedundantErpSl partition Lists.partition(RedundantErpSlList, 1000);
try {CompletableFutureVoid future CompletableFuture.allOf(partition.stream().map(addPartitionList -CompletableFuture.runAsync(() - {//Spring事务内部由ThreadLocal存储事务绑定信息因此需要每个线程新开一个事务DefaultTransactionDefinition defGo new 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(批量导入出现异常, e);//向外抛出异常确保最外面catch回滚throw new PtmException(e.getMessage());});future.join();for (TransactionStatus status : traStatusList) {transactionManager.commit(status);}
} catch (Exception e) {log.error(批量导入出现异常回滚开始, e);for (TransactionStatus status : traStatusList) {transactionManager.rollback(status);}
}先说说这个错误例子我当时开发的思路手动开启事务后在每个线程操作开始的时候都创建一个事务Spring事务传播级别用的TransactionDefinition.PROPAGATION_REQUIRES_NEW即默认创建新事务。隔离级别一开始没改然后我就尝试着操作了一下好家伙批量新增的时候直接锁了。 查看正在锁的事务 SELECT * FROM INFORMATION_SCHEMA.INNODB_LOCKS; 查看等待锁的事务 SELECT * FROM INFORMATION_SCHEMA.INNODB_LOCK_WAITS; 异常如下图锁超时异常 第一次看见下图这个错的时候我是疑惑的没有当回事以为是多数据源的问题。我项目里有直连Oracle和MySQL两种关系型数据库当时怀疑是多数据源事务没有正确解绑导致的问题。 :::info PS事实上这个坑给足了我提示根本原因就是多线程事务解绑失败但是我理解出现了偏差为后文埋下了伏笔。 ::: 我当时一看有锁这我能惯着马上修改事务隔离级别为TransactionDefinition.ISOLATION_READ_COMMITTED读已提交(MySQL默认事务隔离级别为可重复读这里下调了一级总共四级)。顺利插入但还是报上面这个错错误位置是下面这个循环提交时报的第二次循环的时候一定会报错。
for (TransactionStatus status : traStatusList) {transactionManager.commit(status);
}当时一度以为是多数据源的问题但是Debug后发现resource里面只有一个数据源key解绑一次后就没了第二个循环解绑的时候就报上面这个错因为找不到可以解绑的key了。我就很疑惑为啥就一个数据源key我不是在别的线程开了事务嘛按理说开了多少个线程就有多少个事务这个问题困扰了我大概一天左右的时间。然后我想到了Spring事务的实现原理ThreadLocal然后联想到我的多线程开启事务再看到我在主线程里面进行傻叉循环解绑我瞬间为梦想窒息。 所以破案了我在主线程是操作不了子线程事务这也是代码报key找不到的原因因为用主线程做key在ThreadLocal里肯定是拿不到子线程信息的只能拿到主线程自己的。
多线程事务提交方案
因此解决方案就很简单子线程的事务自己操作那么多线程事务处理哪家强JDK里找CompletableFuture当然这里使用CountDownLatch也是可行的网上也有案例。多线程事务在处理逻辑上其实和分布式事务很像因此我这里采用2PC的思想一阶段所有子线程全部开启事务并执行SQL然后阻塞等待二阶段判断是否全部成功是就唤醒所有线程提交事务否就全部回滚。
-----------需要注入Bean一个是Spring Boot事务管理一个是线程池-----------
Autowired
private PlatformTransactionManager transactionManager;
Autowired
Qualifier(ioDenseExecutor)
private ThreadPoolTaskExecutor ioDense;
-----------多线程事务新增操作-----------
private void batchSchedule(ListBudgetBase addList) {if (!CollectionUtils.isEmpty(addList)) {//定义局部变量是否成功、顺序标识、等待线程队列AtomicBoolean isSuccess new AtomicBoolean(true);AtomicInteger cur new AtomicInteger(1);ListThread unfinishedList new ArrayList();//切分新增集合ListListBudgetBase partition Lists.partition(addList, 1000);int totalSize partition.size();//多线程处理开始CompletableFutureVoid future CompletableFuture.allOf(partition.stream().map(addPartitionList - CompletableFuture.runAsync(() - {//Spring事务内部由ThreadLocal存储事务绑定信息因此需要每个线程新开一个事务DefaultTransactionDefinition defGo new DefaultTransactionDefinition();defGo.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);TransactionStatus statusGo transactionManager.getTransaction(defGo);int curInt cur.getAndIncrement();try {log.info(当前是第{}个线程开始启动线程名{}, 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, false);LockSupport.park();if (isSuccess.get()) {log.info(当前是第{}个线程提交线程名{}, curInt, Thread.currentThread().getName());transactionManager.commit(statusGo);} else {log.info(当前是第{}个线程回滚线程名{}, curInt, Thread.currentThread().getName());transactionManager.rollback(statusGo);}} catch (Exception e) {log.error(当前是第{}个线程出现异常线程名{}, curInt, Thread.currentThread().getName(), e);transactionManager.rollback(statusGo);isSuccess.set(false);notifyAllThread(unfinishedList, totalSize, true);}}, ioDense)).toArray(CompletableFuture[]::new));future.join();}}
private void notifyAllThread(ListThread unfinishedList, int totalSize, boolean isForce) {if (isForce || unfinishedList.size() totalSize) {log.info(唤醒当前所有休眠线程线程数{}总线程数{},是否强制{}, unfinishedList.size(), totalSize, isForce);for (Thread thread : unfinishedList) {log.info(当前线程{}被唤醒, thread.getName());LockSupport.unpark(thread);}}}方案详解
为什么用LockSupport的park()和unpark()而不用Thread.sleep()、Object.wait()、Condition.await()?
更简单不需要获取锁能直接阻塞线程。更直观以thread为操作对象更符合阻塞线程的直观定义更精确可以准确地唤醒某一个线程notify随机唤醒一个线程notifyAll唤醒所有等待的线程更灵活 unpark方法可以在park方法前调用。
第4点很重要如果不能提前使用unpark()的话按照代码逻辑最后一个线程会被永久阻塞。
为什么要自建线程池
CompletableFuture默认的线程池ForkJoinPool.commonPool()偏向于计算密集型任务处理核心线程数和逻辑CPU数少1对于多线程事务这种IO密集型任务来说核心线程数偏少。并且上述方法在操作中都是阻塞线程无法一次性开启全部线程的话会导致notifyAllThread方法无法执行老线程阻塞新线程无法开启就尬住了。 ForkJoinPool基于工作窃取算法所以最适合的是计算密集型任务这里我们开启一个参数调整为IO密集型(多核心少队列)的ThreadPoolTaskExecutor线程池即可。
注意MySQL/Druid等数据库的最大连接数
使用多线程的时候也别忘了调整其他组件的最大连接数。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个之后达到上线就会一直卡住原因是老的线程挂起不会释放新的线程因为线程池满了无法创建。因此在使用该方案时一定要估算数据量分好合适的大小连接池和数据库的最大连接数也要注意是否匹配。
