网络公司网站设计维护合同百度网站官网入口网址
某天搬砖时遇到一个问题,我创建了一个线程池执行任务,刚开始的时候还是一切,结果第二天发现有些任务没有正常执行。一看日志才发现是高峰期时线程池给我占用慢了,任务被丢掉了。
举个例子,我创建了一个线程池,最大线程数是10,等待队列最大量是1000,结果高峰期时一下给我来了2000个任务,这个时候自然是顶不住的。解决办法肯定是有的,比如改一下线程池的最大等待队列,扩大到2000以上,但是这种办法肯定不是完美的,如果任务量再次增涨,我又得去改一此代码吗,这样肯定不行。于是我把多余的任务存储到数据库中,弄个定时器去专门执行重跑。
下面是我写的一个例子:
import lombok.Data;import java.util.*;
import java.util.concurrent.*;public class MyTest {// 创建线程池static ThreadPoolExecutor executorService = new ThreadPoolExecutor(2, 3, 2L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());// 创建一个map,假设它是数据库static ConcurrentHashMap<Integer, Integer> dbMap = new ConcurrentHashMap<>();public static void main(String[] args) {// 用户通过正常操作流程来了一批任务executeTasks();// 启动定时任务redoTask();}static void executeTasks() {// 一次性安排30个任务for (int i = 0; i < 30; i++) {// 获取剩余队列的数量int queueCapacity = executorService.getQueue().remainingCapacity();if (queueCapacity < 1) {// 当前队列数不足,记录到数据库dbMap.put(i, i);continue;}System.out.println("当前队列数:" + queueCapacity);executorService.submit(new MyThread(i));}}@Datastatic class MyThread implements Runnable {int taskId;public MyThread(int taskId) {this.taskId = taskId;}@Overridepublic void run() {try {// 模拟:处理一个任务需要5秒TimeUnit.SECONDS.sleep(5);System.out.println("执行任务\"" + taskId + "\"完毕");} catch (InterruptedException e) {e.printStackTrace();}}}/*** 模拟一个定时器,不断扫描数据库中需要重跑的任务* 正式开发肯定网上找一个定时器用起来,我家里穷用不起,就自己随便写一个*/static void redoTask() {while (true) {// 判断数据库中是否有需要重跑的任务if (dbMap.size() > 0) {System.out.println("开始执行重跑任务");int queueCapacity = executorService.getQueue().remainingCapacity();// 当前队列剩余大于一半// 是否是大于一半的时候才运行重跑看具体情况if (queueCapacity > 2) {System.out.println("当前队列资源足够,执行重跑");List<Integer> ids = new LinkedList<>();// 从数据库中取重跑任务dbMap.forEach((k, v) -> {// 因为从数据库里取任务是需要消耗一定时间的,为防止资源再次被占满,再获取一次队列剩余大小int queueCapacity2 = executorService.getQueue().remainingCapacity();// 为避免把资源全部占用,留一点给正常流程来的任务// 因为这里测试开的队列比较小,只留了1,根据实际情况而定if (queueCapacity2 > 1) {// 记录已经重跑过的任务idids.add(k);// 重跑任务executorService.submit(new MyThread(v));} else {// 如果资源占用太多则什么都不执行,等待下一次扫描再执行// 有可能会出现一个问题,某个任务等待了很久也轮不到它,建议按时间排个序}});// 删除已经执行重跑的任务for (Integer id : ids) {dbMap.remove(id);}} else {System.out.println("当前队列资源不足,跳过这次重跑");}}try {// 等待30秒,继续判断TimeUnit.SECONDS.sleep(30);} catch (InterruptedException e) {}}}
}
当然除了入库以外还有其他的办法,我在网上搜索了一下,延迟队列也可以解决: 延迟队列解决方法,不过相比之下我个人更喜欢入库的方案,因为入库之后,每个失败的任务我都能记录日志,方便后期做分析统计。