主要基于小米最近的多线程项目,抽离出里面的多线程实例。
昨天听到袁老去世的消息,非常震惊,一代巨星的陨落,希望袁老在天堂能安好,贴了一张袁老的照片,这张照片也让我想起已在天堂的爷爷。
前言
Java多线程的学习,也有大半个月了,从开始学习Java多线程时,就给自己定了一个小目标,希望能写一个多线程的Demo,今天主要是兑现这个小目标。
这个多线程的示例,其实是结合最近小米的一个多线程异步任务的项目,我把里面涉及到多线程的代码抽离出来,然后进行一定的改造,之所以不自己重写一个,一方面是自己能力还不够,另一方面是想学习现在项目中多线程的实现姿势,至少这个示例是实际项目中应用的。先学习别人怎么造轮子,后面就知道自己怎么去造轮子了。
业务需求
做这个多线程异步任务,主要是因为我们有很多永动的异步任务,什么是永动呢?就是任务跑起来后,需要一直跑下去,比如消息Push任务,因为一直有消息过来,所以需要一直去消费DB中的未推送消息,就需要整一个Push的永动异步任务。
我们的需求其实不难,简单总结一下:
- 能同时执行多个永动的异步任务;
- 每个异步任务,支持开多个线程去消费这个任务的数据;
- 支持永动异步任务的优雅关闭,即关闭后,需要把所有的数据消费完毕后,再关闭。
完成上面的需求,需要注意几个点:
- 每个永动任务,可以开一个线程去执行;
- 每个子任务,因为需要支持并发,需要用线程池控制;
- 永动任务的关闭,需要通知子任务的并发线程,并支持永动任务和并发子任务的优雅关闭。
项目示例
线程池
对于子任务,需要支持并发,如果每个并发都开一个线程,用完就关闭,对资源消耗太大,所以引入线程池:
public class TaskProcessUtil {
// 每个任务,都有自己单独的线程池
private static Map<String, ExecutorService> executors = new ConcurrentHashMap<>();
// 初始化一个线程池
private static ExecutorService init(String poolName, int poolSize) {
return new ThreadPoolExecutor(poolSize, poolSize,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
new ThreadFactoryBuilder().setNameFormat("Pool-" + poolName).setDaemon(false).build(),
new ThreadPoolExecutor.CallerRunsPolicy());
}
// 获取线程池
public static ExecutorService getOrInitExecutors(String poolName,int poolSize) {
ExecutorService executorService = executors.get(poolName);
if (null == executorService) {
synchronized (TaskProcessUtil.class) {
executorService = executors.get(poolName);
if (null == executorService) {
executorService = init(poolName, poolSize);
executors.put(poolName, executorService);
}
}
}
return executorService;
}
// 回收线程资源
public static void releaseExecutors(String poolName) {
ExecutorService executorService = executors.remove(poolName);
if (executorService != null) {
executorService.shutdown();
}
}
}
这是一个线程池的工具类,这里初始化线程池和回收线程资源很简单,我们主要讨论获取线程池。获取线程池可能会存在并发情况,所以需要加一个synchronized锁,然后锁住后,需要对executorService进行二次判空校验,这个和Java单例的实现很像,具体可参考《【设计模式系列5】单例模式》这篇文章。
单个任务
为了更好讲解单个任务的实现方式,我们的任务主要就是把Cat的数据打印出来,Cat定义如下:
@Data
@Service
public class Cat {
private String catName;
public Cat setCatName(String name) {
this.catName = name;
return this;
}
}
单个任务主要包括以下功能:
- 获取永动任务数据:这里一般都是扫描DB,我直接就简单用queryData()代替。
- 多线程执行任务:需要把数据拆分成4份,然后分别由多线程并发执行,这里可以通过线程池支持;
- 永动任务优雅停机:当外面通知任务需要停机,需要执行完剩余任务数据,并回收线程资源,退出任务;
- 永动执行:如果未收到停机命令,任务需要一直执行下去。
直接看代码:
public class ChildTask {
private final int POOL_SIZE = 3; // 线程池大小
private final int SPLIT_SIZE = 4; // 数据拆分大小
private String taskName;
// 接收jvm关闭信号,实现优雅停机
protected volatile boolean terminal = false;
public ChildTask(String taskName) {
this.taskName = taskName;
}
// 程序执行入口
public void doExecute() {
int i = 0;
while(true) {
System.out.println(taskName + ":Cycle-" + i + "-Begin");
// 获取数据
List<Cat> datas = queryData();
// 处理数据
taskExecute(datas);
System.out.println(taskName + ":Cycle-" + i + "-End");
if (terminal) {
// 只有应用关闭,才会走到这里,用于实现优雅的下线
break;
}
i++;
}
// 回收线程池资源
TaskProcessUtil.releaseExecutors(taskName);
}
// 优雅停机
public void terminal() {
回复