今天教大家撸一个 Java 的多线程永动任务,这个示例的原型是公司自研的多线程异步任务项目,我把里面涉及到多线程的代码抽离出来,然后进行一定的改造。

里面涉及的知识点非常多,特别适合有一定工作经验的同学学习,或者可以直接拿到项目中使用。

文章结构非常简单:

1. 功能说明

做这个多线程异步任务,主要是因为我们有很多永动的异步任务,什么是永动呢?就是任务跑起来后,需要一直跑下去。

比如消息 Push 任务,因为一直有消息过来,所以需要一直去消费 DB 中的未推送消息,就需要整一个 Push 的永动异步任务。

我们的需求其实不难,简单总结一下:

  1. 能同时执行多个永动的异步任务

  2. 每个异步任务,支持开多个线程去消费这个任务的数据;

  3. 支持永动异步任务的优雅关闭,即关闭后,需要把所有的数据消费完毕后,再关闭。

完成上面的需求,需要注意几个点:

  1. 每个永动任务,可以开一个线程去执行;

  2. 每个子任务,因为需要支持并发,需要用线程池控制;

  3. 永动任务的关闭,需要通知子任务的并发线程,并支持永动任务和并发子任务的优雅关闭

2. 多线程任务示例

2.1 线程池

对于子任务,需要支持并发,如果每个并发都开一个线程,用完就关闭,对资源消耗太大,所以引入线程池:

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 进行二次判空校验。

2.2 单个任务

为了更好讲解单个任务的实现方式,我们的任务主要就是把 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() {// 关机terminal = true;System.out.println(taskName + " shut down");}// 处理数据private void doProcessData(List<Cat> datas, CountDownLatch latch) {try {for (Cat cat : datas) {System.out.println(taskName + ":" + cat.toString() + ",ThreadName:" + Thread.currentThread().getName());Thread.sleep(1000L);}} catch (Exception e) {System.out.println(e.getStackTrace());} finally {if (latch != null) {latch.countDown();}}}// 处理单个任务数据private void taskExecute(List<Cat> sourceDatas) {if (CollectionUtils.isEmpty(sourceDatas)) {return;}// 将数据拆成4份List<List<Cat>> splitDatas = Lists.partition(sourceDatas, SPLIT_SIZE);final CountDownLatch latch = new CountDownLatch(splitDatas.size());// 并发处理拆分的数据,共用一个线程池for (final List<Cat> datas : splitDatas) {ExecutorService executorService = TaskProcessUtil.getOrInitExecutors(taskName, POOL_SIZE);executorService.submit(new Runnable() {@Overridepublic void run() {doProcessData(datas, latch);}});}try {latch.await();} catch (Exception e) {System.out.println(e.getStackTrace());}}// 获取永动任务数据private List<Cat> queryData() {List<Cat> datas = new ArrayList<>();for (int i = 0; i < 5; i ++) {datas.add(new Cat().setCatName("罗小黑" + i));}return datas;}
}

简单解释一下:

  • queryData:用于获取数据,实际应用中其实是需要把 queryData 定为抽象方法,然后由各个任务实现自己的方法。

  • doProcessData:数据处理逻辑,实际应用中其实是需要把 doProcessData 定为抽象方法,然后由各个任务实现自己的方法。

  • taskExecute:将数据拆分成 4 份,获取该任务的线程池,并交给线程池并发执行,然后通过 latch.await() 阻塞。当这 4 份数据都执行成功后,阻塞结束,该方法才返回。

  • terminal:仅用于接受停机命令,这里该变量定义为 volatile,所以多线程内存可见;

  • doExecute:程序执行入口,封装了每个任务执行的流程,当 terminal=true 时,先执行完任务数据,然后回收线程池,最后退出。

2.3 任务入口

直接上代码:

public class LoopTask {private List<ChildTask> childTasks;public void initLoopTask() {childTasks = new ArrayList();childTasks.add(new ChildTask("childTask1"));childTasks.add(new ChildTask("childTask2"));for (final ChildTask childTask : childTasks) {new Thread(new Runnable() {@Overridepublic void run() {childTask.doExecute();}}).start();}}public void shutdownLoopTask() {if (!CollectionUtils.isEmpty(childTasks)) {for (ChildTask childTask : childTasks) {childTask.terminal();}}}public static void main(String args[]) throws Exception{LoopTask loopTask = new LoopTask();loopTask.initLoopTask();Thread.sleep(5000L);loopTask.shutdownLoopTask();}
}

每个任务都开一个单独的 Thread,这里我初始化了 2 个永动任务,分别为 childTask1 和 childTask2,然后分别执行,后面 Sleep 了 5 秒后,再关闭任务,我们可以看看是否可以按照我们的预期优雅退出。

2.4 结果分析

执行结果如下:

childTask1:Cycle-0-Begin
childTask2:Cycle-0-Begin
childTask1:Cat(catName=罗小黑0),ThreadName:Pool-childTask1
childTask1:Cat(catName=罗小黑4),ThreadName:Pool-childTask1
childTask2:Cat(catName=罗小黑4),ThreadName:Pool-childTask2
childTask2:Cat(catName=罗小黑0),ThreadName:Pool-childTask2
childTask1:Cat(catName=罗小黑1),ThreadName:Pool-childTask1
childTask2:Cat(catName=罗小黑1),ThreadName:Pool-childTask2
childTask2:Cat(catName=罗小黑2),ThreadName:Pool-childTask2
childTask1:Cat(catName=罗小黑2),ThreadName:Pool-childTask1
childTask2:Cat(catName=罗小黑3),ThreadName:Pool-childTask2
childTask1:Cat(catName=罗小黑3),ThreadName:Pool-childTask1
childTask2:Cycle-0-End
childTask2:Cycle-1-Begin
childTask1:Cycle-0-End
childTask1:Cycle-1-Begin
childTask2:Cat(catName=罗小黑0),ThreadName:Pool-childTask2
childTask2:Cat(catName=罗小黑4),ThreadName:Pool-childTask2
childTask1:Cat(catName=罗小黑4),ThreadName:Pool-childTask1
childTask1:Cat(catName=罗小黑0),ThreadName:Pool-childTask1
childTask1 shut down
childTask2 shut down
childTask2:Cat(catName=罗小黑1),ThreadName:Pool-childTask2
childTask1:Cat(catName=罗小黑1),ThreadName:Pool-childTask1
childTask1:Cat(catName=罗小黑2),ThreadName:Pool-childTask1
childTask2:Cat(catName=罗小黑2),ThreadName:Pool-childTask2
childTask1:Cat(catName=罗小黑3),ThreadName:Pool-childTask1
childTask2:Cat(catName=罗小黑3),ThreadName:Pool-childTask2
childTask1:Cycle-1-End
childTask2:Cycle-1-End

输出数据:

  • “Pool-childTask” 是线程池名称;

  • “childTask” 是任务名称;

  • “Cat(catName=罗小黑)” 是执行的结果;

  • “childTask shut down” 是关闭标记;

  • “childTask:Cycle-X-Begin” 和“childTask:Cycle-X-End” 是每一轮循环的开始和结束标记。

我们分析一下执行结果:

  • childTask1 和 childTask2 分别执行,在第一轮循环中都正常输出了 5 条罗小黑数据;

  • 第二轮执行过程中,我启动了关闭指令,这次第二轮执行没有直接停止,而是先执行完任务中的数据,再执行退出,所以完全符合我们的优雅退出结论。

2.5 源码地址

GitHub 地址:

https://github.com/lml200701158/java-study/tree/master/src/main/java/com/java/parallel/pool/ofc

新来个技术总监,仅花2小时,撸出一个多线程永动任务,看完直接跪了,真牛逼!相关推荐

  1. 新来个阿里 P7,仅花 2 小时,做出一个多线程永动任务,看完直接跪了

    今天教大家做一个 Java 的多线程永动任务,这个示例的原型是公司自研的多线程异步任务项目,我把里面涉及到多线程的代码抽离出来,然后进行一定的改造. 里面涉及的知识点非常多,特别适合有一定工作经验的同 ...

  2. 新来个阿里 P7,仅花 2 小时,撸出一个多线程永动任务,看完直接跪了,真牛逼

    大家好,我是楼仔! 今天教大家撸一个 Java 的多线程永动任务,这个示例的原型是公司自研的多线程异步任务项目,我把里面涉及到多线程的代码抽离出来,然后进行一定的改造. 里面涉及的知识点非常多,特别适 ...

  3. 新来个阿里 P7,仅花 2 小时,撸出一个多线程永动任务,看完直接跪了,真牛逼!...

    大家好,今天教大家撸一个 Java 的多线程永动任务,这个示例的原型是公司自研的多线程异步任务项目,我把里面涉及到多线程的代码抽离出来,然后进行一定的改造. 里面涉及的知识点非常多,特别适合有一定工作 ...

  4. 一种新的计算机视觉技术?将手机的摄像头变成了一个搜索引擎

    https://www.toutiao.com/a6698870195391300099/ 智能手机,安全摄像头和扬声器仅仅是行将运行更加人工智能软件以加快图像和语音处置使命的装备中的一小部分.增添硬 ...

  5. 趣店新项目万里目,百亿补贴计划,能烧出一个未来吗

    一个做消费金融的企业,转行去做奢侈品跨境电商.从百亿美元市值,到市值不足5亿美金,身处退市边缘的趣店,又开始"花式"转型了. 这次被盯上的是"奢侈品电商",推出 ...

  6. 去了家新公司,技术总监不让用 IntelliJ IDEA!!想离职了。。

    最近有个小伙伴微信和我说,新去的一家公司,技术团队全部规定要用的 Eclipse 开发,技术总监不让用 IntelliJ IDEA,付费也不行,说想离职了,问我该怎么办. 首先听到这件事情的时候,我表 ...

  7. 去了家新公司,技术总监不让用 IntelliJ IDEA想离职了

    最近有个小伙伴微信和我说,新去的一家公司,技术团队全部规定要用的 Eclipse 开发,技术总监不让用 IntelliJ IDEA,付费也不行,说想离职了,问我该怎么办. 首先听到这件事情的时候,我表 ...

  8. 去了家新公司,技术总监不让用 IntelliJ IDEA 想离职了

    最近有个小伙伴微信和我说,新去的一家公司,技术团队全部规定要用的 Eclipse 开发,技术总监不让用 IntelliJ IDEA,付费也不行,说想离职了,问我该怎么办. 首先听到这件事情的时候,我表 ...

  9. 去了家新公司,技术总监不让用 IntelliJ IDEA,想离职了。。

    最近有个小伙伴微信和我说,新去的一家公司,技术团队全部规定要用的 Eclipse 开发,技术总监不让用 IntelliJ IDEA,付费也不行,说想离职了,问我该怎么办. 首先听到这件事情的时候,我表 ...

最新文章

  1. 全球及中国智能音箱市场规模产值及发展机遇研究报告2021-2027年
  2. Spark算子总结版
  3. how is metadata got - DB table iwfndi_med_srh and IWFNDCL_MGW_REQUEST_MANAG
  4. 管理者的困境:放权或者崩溃
  5. 你喜欢那种类型的收集壁纸?
  6. Messari:自2019年,DeFi领域因黑客攻击损失超2.84亿美元资产
  7. 基于Yolov5目标检测的物体分类识别及定位(二) -- yolov5运行环境搭建及label格式转换
  8. mysql数据库软件 国产_国产数据库发展情况如何?
  9. 消息模板取数据的高阶使用说明
  10. 想要挣钱创收 那就用脚本操作手机群控软件啊
  11. 今日头条笔试8/23第二题
  12. android ps icon图标制作,PS设计App图标教程
  13. blender反选快捷键_【PS】常用操作及快捷键
  14. 计算机设置桌面文件夹,win10电脑怎么更改桌面文件夹路径
  15. python打开setting_Python3 - setting的默认配置和用户配置读取
  16. 致远OA漏洞分析、利用与防护合集
  17. linux中.sh文件是什么?怎么执行?
  18. 接连入职五位博士,云和恩墨到底在搞什么大事情?
  19. 派大汤的数据结构错题本
  20. chrome浏览器中使用adblockplus拦截广告

热门文章

  1. Bootstrap 轻松实现选项卡
  2. 《奔跑吧Linux内核》开始预售啦
  3. 安装openssh-server报Depends: openssh-client (= 1:6.6p1-2ubuntu2.8)错误
  4. Google谷歌权重09年算法
  5. 社区计算机培训班总结,2019年社区计算机培训工作总结范文
  6. linux的备份和恢复命令,Ubuntu 16.04备份和恢复小结
  7. AlertDialog基本使用方法
  8. 数据库设计中各种键的含义
  9. 二分查找以及其有趣的使用
  10. 软件的注册码及清除电脑垃圾的文件