一级域名做网站中国站长之家官网
某天搬砖时遇到一个问题，我创建了一个线程池执行任务，刚开始的时候还是一切，结果第二天发现有些任务没有正常执行。一看日志才发现是高峰期时线程池给我占用慢了，任务被丢掉了。
举个例子，我创建了一个线程池，最大线程数是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) {}}}
}
当然除了入库以外还有其他的办法，我在网上搜索了一下，延迟队列也可以解决： 延迟队列解决方法，不过相比之下我个人更喜欢入库的方案，因为入库之后，每个失败的任务我都能记录日志，方便后期做分析统计。