Java--使用ExecutorCompletionService 管理线程池处理任务的返回结果

快来打我* 2022-05-30 02:28 170阅读 0赞

在我们日常使用线程池的时候，经常会有需要获得线程处理结果的时候。此时我们通常有两种做法。

1. 使用并发容器将callable.call() 的返回Future存储起来。然后使用一个消费者线程去遍历这个并发容器，调用Future.isDone()去判断各个任务是否处理完毕。然后再处理响应的业务。

![复制代码][copycode.gif]

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class ExecutorResultManager {
    
        public static void main(String[] args) {
            // 队列
            BlockingQueue<Future<String>> futures = new LinkedBlockingQueue<>();
            
            // 生产者
            new Thread() {
                @Override
                public void run() {
                    ExecutorService pool = Executors.newCachedThreadPool();
                    
                    for (int i=0; i< 10; i++) {
                        int index = i;
                        Future<String> submit = pool.submit(new Callable<String>() {
                            @Override
                            public String call() throws Exception {
                                return "task done" + index;
                            }
                        });
                        try {
                            futures.put(submit);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }.start();
            
            
    
            // 消费者
            new Thread() {
                @Override
                public void run() {
                    while(true) {
                        for (Future<String> future : futures) {
                            if(future.isDone()) {
                                // 处理业务
                                // .............
                                
                                
                            };
                        }
                    }
                }
            }.start();
        }
    
    }

![复制代码][copycode.gif]

2. 使用jdk 自带线程池结果管理器：ExecutorCompletionService。它将BlockingQueue 和Executor 封装起来。然后使用ExecutorCompletionService.submit()方法提交任务。

submit 方法如下：

![复制代码][copycode.gif]

public Future<V> submit(Callable<V> task) {
            if (task == null) throw new NullPointerException();
     
            // RunnableFuture封装了任务，使得任务既能run 也能get()
            RunnableFuture<V> f = newTaskFor(task);
            // 使用一个继承Runnable类的QueueingFutue 再次封装了我们的任务
            executor.execute(new QueueingFuture(f));
            return f;
       }

![复制代码][copycode.gif]

我们再来看看QueueingFuture：

![复制代码][copycode.gif]

// 继承自FutureTask, FutureTask 也是一个Runnable的子类
     private class QueueingFuture extends FutureTask<Void> {
            QueueingFuture(RunnableFuture<V> task) {
                super(task, null);
                this.task = task;
            }
            // 实现了FutureTask 的done 方法，在任务处理完毕或者抛异常后将封装成Future的任务加入到队列。这样我们就能在队列中取到处处理完的任务，并通过Future.get()方法去取得处理完后的结果。不用自己去判断任务是否处理完毕了
            protected void done() { completionQueue.add(task); }
            private final Future<V> task;
        }

![复制代码][copycode.gif]

简单实现：

![复制代码][copycode.gif]

import java.util.concurrent.Callable;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    public class ExecutorCompletionServiceManager {
    
        public static void main(String[] args) {
    
            ExecutorCompletionService<String> service = new ExecutorCompletionService<String>(
                    Executors.newCachedThreadPool());
    
            // 生产者
            new Thread() {
                @Override
                public void run() {
                    for (int i = 0; i < 10; i++) {
                        int index = i;
                        service.submit(new Callable<String>() {
                            @Override
                            public String call() throws Exception {
                                return "task done" + index;
                            }
                        });
                    }
                }
            }.start();
            
    
            // 消费者
            new Thread() {
                @Override
                public void run() {
                    try {
                        Future<String> take = service.take();
                        // do some thing........
                        
                        
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
    
        }
    }

![复制代码][copycode.gif]

相对于原始的自造轮子处理方式，jdk 自带的工具类处理方式显得优雅了许多。

[copycode.gif]: /images/20220530/602bcd35c7c749bcaf71585633e66be8.png