4.线程安全问题
既然使用了线程,伴随而来的还会有线程安全问题。
假如现在有这样一个需求:用多线程执行查询方法,然后把执行结果添加到一个list集合中。
代码如下:
List<User> list = Lists.newArrayList(); dataList.stream() .map(data -> CompletableFuture .supplyAsync(() -> query(list, data), asyncExecutor) )); CompletableFuture.allOf(futureArray).join();
使用CompletableFuture
异步多线程执行query方法:
public void query(List<User> list, UserEntity condition) { User user = queryByCondition(condition); if(Objects.isNull(user)) { return; } list.add(user); UserExtend userExtend = queryByOther(condition); if(Objects.nonNull(userExtend)) { user.setExtend(userExtend.getInfo()); } }
在query方法中,将获取的查询结果添加到list集合中。
结果list会出现线程安全问题,有时候会少数据,当然也不一定是必现的。
这是因为ArrayList
是非线程安全
的,没有使用synchronized
等关键字修饰。
如何解决这个问题呢?
答:使用CopyOnWriteArrayList
集合,代替普通的ArrayList
集合,CopyOnWriteArrayList是一个线程安全的机会。
只需一行小小的改动即可:
List<User> list Lists.newCopyOnWriteArrayList();
温馨的提醒一下,这里创建集合的方式,用了google的collect包。
5.ThreadLocal获取数据异常
我们都知道JDK
为了解决线程安全问题,提供了一种用空间换时间的新思路:ThreadLocal
。
它的核心思想是:共享变量在每个线程
都有一个副本
,每个线程操作的都是自己的副本,对另外的线程没有影响。
例如:
@Service public class ThreadLocalService { private static final ThreadLocal<Integer> threadLocal = new ThreadLocal<>(); public void add() { threadLocal.set(1); doSamething(); Integer integer = threadLocal.get(); } }
ThreadLocal在普通中线程中,的确能够获取正确的数据。
但在真实的业务场景中,一般很少用单独的线程
,绝大多数,都是用的线程池
。
那么,在线程池中如何获取ThreadLocal
对象生成的数据呢?
如果直接使用普通ThreadLocal,显然是获取不到正确数据的。
我们先试试InheritableThreadLocal
,具体代码如下:
private static void fun1() { InheritableThreadLocal<Integer> threadLocal = new InheritableThreadLocal<>(); threadLocal.set(6); System.out.println("父线程获取数据:" + threadLocal.get()); ExecutorService executorService = Executors.newSingleThreadExecutor(); threadLocal.set(6); executorService.submit(() -> { System.out.println("第一次从线程池中获取数据:" + threadLocal.get()); }); threadLocal.set(7); executorService.submit(() -> { System.out.println("第二次从线程池中获取数据:" + threadLocal.get()); }); }
执行结果:
父线程获取数据:6 第一次从线程池中获取数据:6 第二次从线程池中获取数据:6
由于这个例子中使用了单例线程池,固定线程数是1。
第一次submit任务的时候,该线程池会自动创建一个线程。因为使用了InheritableThreadLocal,所以创建线程时,会调用它的init方法,将父线程中的inheritableThreadLocals数据复制到子线程中。所以我们看到,在主线程中将数据设置成6,第一次从线程池中获取了正确的数据6。
之后,在主线程中又将数据改成7,但在第二次从线程池中获取数据却依然是6。
因为第二次submit任务的时候,线程池中已经有一个线程了,就直接拿过来复用,不会再重新创建线程了。所以不会再调用线程的init方法,所以第二次其实没有获取到最新的数据7,还是获取的老数据6。
那么,这该怎么办呢?
答:使用TransmittableThreadLocal
,它并非JDK自带的类,而是阿里巴巴开源jar包中的类。
可以通过如下pom文件引入该jar包:
<dependency> <groupId>com.alibaba</groupId> <artifactId>transmittable-thread-local</artifactId> <version>2.11.0</version> <scope>compile</scope> </dependency>
代码调整如下:
private static void fun2() throws Exception { TransmittableThreadLocal<Integer> threadLocal = new TransmittableThreadLocal<>(); threadLocal.set(6); System.out.println("父线程获取数据:" + threadLocal.get()); ExecutorService ttlExecutorService = TtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(1)); threadLocal.set(6); ttlExecutorService.submit(() -> { System.out.println("第一次从线程池中获取数据:" + threadLocal.get()); }); threadLocal.set(7); ttlExecutorService.submit(() -> { System.out.println("第二次从线程池中获取数据:" + threadLocal.get()); }); }
执行结果:
父线程获取数据:6 第一次从线程池中获取数据:6 第二次从线程池中获取数据:7
我们看到,使用了TransmittableThreadLocal之后,第二次从线程中也能正确获取最新的数据7了。
nice。
如果你仔细观察这个例子,你可能会发现,代码中除了使用TransmittableThreadLocal
类之外,还使用了TtlExecutors.getTtlExecutorService
方法,去创建ExecutorService
对象。
这是非常重要的地方,如果没有这一步,TransmittableThreadLocal
在线程池中共享数据将不会起作用。
创建ExecutorService
对象,底层的submit方法会TtlRunnable
或TtlCallable
对象。
以TtlRunnable类为例,它实现了Runnable
接口,同时还实现了它的run方法:
public void run() { Map<TransmittableThreadLocal<?>, Object> copied = (Map)this.copiedRef.get(); if (copied != null && (!this.releaseTtlValueReferenceAfterRun || this.copiedRef.compareAndSet(copied, (Object)null))) { Map backup = TransmittableThreadLocal.backupAndSetToCopied(copied); try { this.runnable.run(); } finally { TransmittableThreadLocal.restoreBackup(backup); } } else { throw new IllegalStateException("TTL value reference is released after run!"); } }
这段代码的主要逻辑如下:
- 把当时的ThreadLocal做个备份,然后将父类的ThreadLocal拷贝过来。
- 执行真正的run方法,可以获取到父类最新的ThreadLocal数据。
- 从备份的数据中,恢复当时的ThreadLocal数据。
6.OOM问题
众所周知,使用多线程可以提升代码执行效率,但也不是绝对的。
对于一些耗时的操作,使用多线程,确实可以提升代码执行效率。
但线程不是创建越多越好,如果线程创建多了,也可能会导致OOM
异常。
例如:
Caused by: java.lang.OutOfMemoryError: unable to create new native thread
在JVM
中创建一个线程,默认需要占用1M
的内存空间。
如果创建了过多的线程,必然会导致内存空间不足,从而出现OOM异常。
除此之外,如果使用线程池的话,特别是使用固定大小线程池,即使用Executors.newFixedThreadPool
方法创建的线程池。
该线程池的核心线程数
和最大线程数
是一样的,是一个固定值,而存放消息的队列是LinkedBlockingQueue
。
该队列的最大容量是Integer.MAX_VALUE
,也就是说如果使用固定大小线程池,存放了太多的任务,有可能也会导致OOM异常。
java.lang.OutOfMemeryError:Java heap space
7.CPU使用率飙高
不知道你有没有做过excel数据导入功能,需要将一批excel的数据导入到系统中。
每条数据都有些业务逻辑,如果单线程导入所有的数据,导入效率会非常低。
于是改成了多线程导入。
如果excel中有大量的数据,很可能会出现CPU使用率飙高的问题。
我们都知道,如果代码出现死循环,cpu使用率会飚的很多高。因为代码一直在某个线程中循环,没法切换到其他线程,cpu一直被占用着,所以会导致cpu使用率一直高居不下。
而多线程导入大量的数据,虽说没有死循环代码,但由于多个线程一直在不停的处理数据,导致占用了cpu很长的时间。
也会出现cpu使用率很高的问题。
那么,如何解决这个问题呢?
答:使用Thread.sleep
休眠一下。
在线程中处理完一条数据,休眠10毫秒。
当然CPU使用率飙高的原因很多,多线程处理数据和死循环只是其中两种,还有比如:频繁GC、正则匹配、频繁序列化和反序列化等。
后面我会写一篇介绍CPU使用率飙高的原因的专题文章,感兴趣的小伙伴,可以关注一下我后续的文章。
8.事务问题
在实际项目开发中,多线程的使用场景还是挺多的。如果spring事务用在多线程场景中,会有问题吗?
例如:
@Slf4j @Service public class UserService { @Autowired private UserMapper userMapper; @Autowired private RoleService roleService; @Transactional public void add(UserModel userModel) throws Exception { userMapper.insertUser(userModel); new Thread(() -> { roleService.doOtherThing(); }).start(); } } @Service public class RoleService { @Transactional public void doOtherThing() { System.out.println("保存role表数据"); } }
从上面的例子中,我们可以看到事务方法
add中,调用了事务方法doOtherThing,但是事务方法
doOtherThing是在另外一个线程
中调用的。
这样会导致两个方法不在同一个线程中,获取到的数据库连接
不一样,从而是两个不同的事务
。如果想doOtherThing方法中抛了异常,add方法也回滚是不可能的。
如果看过spring事务源码的朋友,可能会知道spring的事务是通过数据库连接来实现的。当前线程中保存了一个map,key是数据源
,value是数据库连接
。
private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>("Transactional resources");
我们说的同一个事务
,其实是指同一个数据库连接
,只有拥有同一个数据库连接才能同时提交
和回滚
。如果在不同的线程
,拿到的数据库连接
肯定是不一样的,所以是不同的事务。
所以不要在事务中开启另外的线程,去处理业务逻辑,这样会导致事务失效。
9.导致服务挂掉
使用多线程会导致服务挂掉,这不是危言耸听,而是确有其事。
假设现在有这样一种业务场景:在mq的消费者中需要调用订单查询接口,查到数据之后,写入业务表中。
本来是没啥问题的。
突然有一天,mq生产者跑了一个批量数据处理的job,导致mq服务器上堆积了大量的消息。
此时,mq消费者的处理速度,远远跟不上mq消息的生产速度,导致的结果是出现了大量的消息堆积,对用户有很大的影响。
为了解决这个问题,mq消费者改成多线程
处理,直接使用了线程池
,并且最大线程数
配置成了20。
这样调整之后,消息堆积问题确实得到了解决。
但带来了另外一个更严重的问题:订单查询接口并发量太大了,有点扛不住压力,导致部分节点的服务直接挂掉。
为了解决问题,不得不临时加服务节点。
在mq的消费者中使用多线程,调用接口时,一定要评估好接口能够承受的最大访问量,防止因为压力过大,而导致服务挂掉的问题。