CompletionService按序获取线程结果

爱被打了一巴掌 2023-01-22 13:50 26阅读 0赞

### 文章目录 ###

*   *  使用场景
     *  原理
     *  api
     *  使用
     *   *  场景1：提交了多个任务，但只要有一个任务完成并返回一个非空的结果，并可以忽略掉其余的任务
         *  场景2：提交了多个任务，按执行完成顺序获取结果
     *  传统做法弊端

## 使用场景 ##

*  多个线程，先执行完的进阻塞队列，然后可以按执行顺序获取结果
 *  提交了多个任务，但只要有一个任务完成并返回一个非空的结果，忽略掉其余的任务

## 原理 ##

内部维护了一个阻塞队列，提交的任务，先执行完的先进入队列，所以获取值肯定是按：执行完成的顺序。

## api ##

Future submit(Callable task);
    submit 用于提交一个 Callable 对象，用于提交一个可以获得结果的线程任务

Future submit(Runnable task, V result);
    submit 用于提交一个 Runnable 对象及 result 对象，类似于上面的 submit，但是 runnable 的返回值 void 无法获得线程的结果，所以添加了 result 用于做为参数的桥梁

Future take() throws InterruptedException;
    take 用于取出最新的线程执行结果，注意这里是阻塞的

Future poll();
    take 用于取出最新的线程执行结果，是非阻塞的，如果没有结果就返回 null

Future poll(long timeout, TimeUnit unit) throws InterruptedException;
    同上，只是加了一个超时时间

## 使用 ##

### 场景1：提交了多个任务，但只要有一个任务完成并返回一个非空的结果，并可以忽略掉其余的任务 ###

void solve(Executor e, Collection<Callable<Result>> solvers) throws InterruptedException {
        
         CompletionService<Result> completionService = new ExecutorCompletionService<Result>(e);
         int n = solvers.size();
         List<Future<Result>> futures
             = new ArrayList<Future<Result>>(n);
         Result result = null;
         try {
        
             //提交多个任务
             for (Callable<Result> s : solvers)
                 futures.add(completionService.submit(s));
            //
             for (int i = 0; i < n; ++i) {
        
                 try {
        
                     //等待获取一个已经完成的任务
                     Result r = completionService.take().get();
                     //判断返回结果是否为空
                     if (r != null) {
        
                         result = r;
                         break;
                     }
                 } catch (ExecutionException ignore) {
        }
             }
         }
         finally {
        
             //取消所有任务
             for (Future<Result> f : futures)
                 f.cancel(true);
         }
    
         if (result != null)
             use(result);
     }

### 场景2：提交了多个任务，按执行完成顺序获取结果 ###

import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.RandomUtils;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorCompletionService;
    
    /**
     * 按序获取线程值
     */
    @Slf4j
    @SpringBootTest
    class CompletionServiceTest {
        
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @Test
        public void test() {
        
    
            CompletionService<Object> completionService = new ExecutorCompletionService<>(threadPoolTaskExecutor);
            for (int i = 0; i < 5; i++) {
        
                completionService.submit(() -> {
        
                    int y = RandomUtils.nextInt(0, 5);
                    System.out.println("当前线程：" + Thread.currentThread().getName() + ",值：" + y);
                    return y;
                });
            }
    
            for (int i = 0; i < 5; i++) {
        
                try {
        
                    System.out.println(completionService.take().get());
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                } catch (ExecutionException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
    }

最后看下传统做法的弊端

## 传统做法弊端 ##

public class CompletionServiceTest {
        
     
        static class Task implements Callable<String>{
        
            private int i;
             
            public Task(int i){
        
                this.i = i;
            }
     
            @Override
            public String call() throws Exception {
        
                Thread.sleep(10000);
                return Thread.currentThread().getName() + "执行完任务：" + i;
            }  
        }
         
        public static void main(String[] args){
        
            testUseFuture();
        }
         
        private static void testUseFuture(){
        
            int numThread = 5;
            ExecutorService executor = Executors.newFixedThreadPool(numThread);
            List<Future<String>> futureList = new ArrayList<Future<String>>();
            for(int i = 0;i<numThread;i++ ){
        
                Future<String> future = executor.submit(new CompletionServiceTest.Task(i));
                futureList.add(future);
            }
                     
            while(numThread > 0){
        
                for(Future<String> future : futureList){
        
                    String result = null;
                    try {
        
                        result = future.get(0, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    } catch (ExecutionException e) {
        
                        e.printStackTrace();
                    } catch (TimeoutException e) {
        
                        //超时异常直接忽略
                    }
                    if(null != result){
        
                        futureList.remove(future);
                        numThread--;
                        System.out.println(result);
                        //此处必须break，否则会抛出并发修改异常。（也可以通过将futureList声明为CopyOnWriteArrayList类型解决）
                        break;
                    }
                }
            }
        }
    }