封装CompletionService的并发任务分发器(优化版)

这个框架代码用了很长时间，使用场景也挺多，初衷是简化CompletionService的编程接口，尽量减少业务代码处的感知。
今天找deepseek做了一版优化，优化点：

• 整体的超时控制
  
• 超时、异常处理和封装
  
• 取消未完成的任务
  

核心代码

1. public class TaskDispatcher<T> {
3.     private final CompletionService<T> completionService;
5.     /**
6.      * 待处理任务
7.      */
8.     private final Set<Future<T>> pending = Sets.newHashSet();
10.     /**
11.      * 超时时间, 单位: s
12.      */
13.     private long timeout = 10000;
15.     public TaskDispatcher(Executor executor, long timeout) {
16.         completionService = new ExecutorCompletionService<>(executor);
17.         if (timeout > 0) {
18.             this.timeout = timeout;
19.         }
20.     }
22.     public void submit(Callable<T> task) {
23.         Future<T> future = completionService.submit(task);
24.         pending.add(future);
25.     }
27.     /**
28.      * 仅获取执行的任务结果
29.      *
30.      * @param ignoreException 忽略执行时发生的异常
31.      * @return
32.      */
33.     public List<T> taskCompletedResult(boolean ignoreException) {
34.         List<TaskResult<T>> taskResultList = taskCompleted();
35.         List<T> res = Lists.newArrayList();
36.         if (CollectionUtils.isEmpty(taskResultList)) {
37.             return res;
38.         }
39.         boolean hasError = false;
40.         for (TaskResult<T> taskResult : taskResultList) {
41.             if (!taskResult.isTimeout() && taskResult.getError() == null) {
42.                 res.add(taskResult.getValue());
43.             } else if (taskResult.isTimeout() && !ignoreException) {
44.                 LoggerUtils.error("执行任务时超时");
45.                 hasError = true;
46.             } else if (taskResult.getError() != null && !ignoreException) {
47.                 LoggerUtils.error("执行任务时发生异常", taskResult.getError());
48.                 hasError = true;
49.             }
50.         }
51.         if (hasError) {
52.             throw new ZHException("任务并发处理时发生异常");
53.         }
54.         return res;
55.     }
57.     /**
58.      * 获取执行的任务
59.      *
60.      * @return
61.      */
62.     public List<TaskResult<T>> taskCompleted() {
63.         long deadline = System.currentTimeMillis() + timeout;
64.         List<TaskResult<T>> results = Lists.newArrayList();
65.         int totalTasks = pending.size();
67.         try {
68.             for (int i = 0; i < totalTasks; i++) {
69.                 long remaining = Math.max(0, deadline - System.currentTimeMillis());
70.                 Future<T> future = completionService.poll(remaining, TimeUnit.MILLISECONDS);
71.                 TaskResult<T> result = new TaskResult<>();
72.                 if (future == null) {
73.                     result.setTimeout(true);
74.                 } else {
75.                     pending.remove(future);
76.                     try {
77.                         result.setValue(future.get());
78.                     } catch (ExecutionException e) {
79.                         result.setError(e.getCause());
80.                     }
81.                 }
82.                 results.add(result);
83.             }
84.         } catch (InterruptedException e) {
85.             Thread.currentThread().interrupt();
86.             throw new RuntimeException("任务结果收集中断", e);
87.         } finally {
88.             pending.forEach(f -> f.cancel(true));
89.             pending.clear();
90.         }
91.         return results;
92.     }
94.     @Data
95.     static class TaskResult<T> {
96.         private T value;
97.         private Throwable error;
98.         private boolean isTimeout;
99.     }
100. }

复制代码

需要自己声明线程池bean，使用方式如下

1.         TaskDispatcher<Integer> taskDispatcher = new TaskDispatcher<Integer>(threadExecutor, TIME_OUT);
2.         for (long index: indexList) {
3.             taskDispatcher.submit(() -> xxxService.count(index));
4.         }

复制代码

为了便于在计数求和场景使用，进一步实现了一个子类

1. public class IntSumTaskDispatcher extends TaskDispatcher<Integer> {
2.     public IntSumTaskDispatcher(Executor executor, long timeout, boolean throwException) {
3.         super(executor, timeout);
4.     }
6.     /**
7.      * 对所有结果求和
8.      *
9.      * @return
10.      */
11.     public int takeCompletedSum() {
12.         List<Integer> countResList = taskCompletedResult(true);
13.         int count = 0;
14.         for (Integer countSingle : countResList) {
15.             if (countSingle == null) {
16.                 continue;
17.             }
18.             count += countSingle;
19.         }
20.         return count;
21.     }
22. }

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！