【Thread】并行多线程任务的创建

柔光的暖阳◎ 2023-10-02 14:09 68阅读 0赞

对于简单的并行任务，可以通过线程池+Future的方案来 解决；如果任务之间有聚合关系，无论是 AND 聚合还是 OR 聚合，都可以通过CompletableFuture来解决；而批量的并行任务，则可以通过CompletionService来解决

### 1. 线程池 + `Future` ###

public class ExampleTwo {
        
    
        public static void main(String[] args) throws Exception {
        
            System.out.println("【主线程】开始执行");
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            // 任务1
            Future<String> f1 = executorService.submit(() -> {
        
                try {
        
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                return "Hello Java";
            });
            // 任务2
            Future<String> f2 = executorService.submit(() -> "Hello Python");
            System.out.println("【主线程】执行中");
            // 主线程会阻塞在这，直至子线程有返回结果
            System.out.println(f1.get());
            System.out.println(f2.get());
            executorService.shutdown();
            System.out.println("【主线程】结束执行");
        }
    
    }

对于上述写法（一个一个提交任务），可以有另外一种，把任务放在集合中（这种情况：每个任务的返回值类型必须一致！！）

public class ExampleOne {
        
    
        public static void main(String[] args) throws Exception {
        
            System.out.println("【主线程】开始执行");
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            List<Callable<String>> callables = new ArrayList<>();
            callables.add(() -> {
        
                try {
        
                    Thread.sleep(2000);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
                return "Hello Java";
            });
            callables.add(() -> "Hello C++");
            callables.add(() -> "Hello Python");
    
            1.invokeAll()/
            // 这里会阻塞等待，等待futures获取到所有异步执行的结果才会执行
            List<Future<String>> futures = executorService.invokeAll(callables);
            for(Future<String> stringFuture : futures) {
        
                System.out.println("future task: " + stringFuture.get());
            }
    
            /2.invokeAny()
            String future = executorService.invokeAny(callables);
            System.out.println(future);
            executorService.shutdown();
            System.out.println("【主线程】结束运行");
        }
    }

说明

1.  将 3 个任务存放在一个集合中；
2.  `invokeAll()` 方法是一并提交，但是会同步等待这些任务的返回结果
3.  `invokeAny`：如果一系列请求，我们并不需要等待每个请求，我们可以用 `invokeAny()` 方法，只要某一个请求返回即可

### 2. 线程池 + `FutureTask` ###

public class ExampleThree {
        
    
        public static void main(String[] args) throws Exception {
        
            System.out.println("【主线程】开始执行");
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            FutureTask<Integer> f1 = new FutureTask<>(() -> {
        
                try {
        
                    Thread.sleep(2000);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
                return 666;
            });
            FutureTask<String> f2 = new FutureTask<>(() -> "Hello Java");
            executorService.submit(f1);
            executorService.submit(f2);
            Integer result1 = f1.get();
            String result2 = f2.get();
            System.out.println(result1);
            System.out.println(result2);
        }
        
    }

`FutureTask` 与 `Future` 关系结构图

public class FutureTask<V> implements RunnableFuture<V> {
        
    	private Callable<V> callable;
    	
    	public FutureTask(Callable<V> callable) {
        
            if (callable == null)
                throw new NullPointerException();
            this.callable = callable;
            this.state = NEW;
        }
    }
    
    public interface RunnableFuture<V> extends Runnable, Future<V> {
        
        void run();
    }

`FutureTask` 类 实现了 `Future` 接口

#### 3. 线程池 + `CompletableFuture` ####

public class ExampleFour {
        
    
        public static void main(String[] args) {
        
            // 1.起10个线程的线程池
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            Long startTIME = System.currentTimeMillis();
    
            //1.runAsync()/
            // 任务1：无返回值
            CompletableFuture<Void> f1 = CompletableFuture.runAsync(() -> {
        
                try {
        
                    Thread.sleep(1000);
                    Long endTIME = System.currentTimeMillis();
                    Long elapsedTime = endTIME - startTIME;
                    System.out.println("任务一 elapsedTime = " + elapsedTime);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }, executorService);
    
            //2.supplyAsync()/
            // 任务2：有返回值；与任务1并行
            CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
        
                Integer result2 = 999;
                try {
        
                    Thread.sleep(2000);
                    Long endTIME = System.currentTimeMillis();
                    Long elapsedTime = endTIME - startTIME;
                    System.out.println("任务二 elapsedTime = " + elapsedTime + "; result2 = " + result2);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
                return result2;
            }, executorService);
    
            // 任务3：有返回值；与任务1并行
            CompletableFuture<Integer> f3 = CompletableFuture.supplyAsync(() -> {
        
                Integer result3 = 666;
                try {
        
                    Thread.sleep(3000);
                    Long endTIME = System.currentTimeMillis();
                    Long elapsedTime = endTIME - startTIME;
                    System.out.println("任务三 elapsedTime = " + elapsedTime + "; result3 = " + result3);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
                return result3;
            }, executorService);
    
            //4.汇聚任务/
            // 任务四 （AND汇聚）： 任务一 和 任务二都完成后才开始
            CompletableFuture<Integer> f4 = f2.thenCombine(f1, (result2,__)->{
        
                Integer result4 = result2;
                Long endTIME = System.currentTimeMillis();
                Long elapsedTime = endTIME - startTIME;
                System.out.println("任务四 elapsedTime = " + elapsedTime);
                return result4;
            });
    
            /* 任务五 （AND汇聚）： 任务二 和 任务三都完成后才开始 */
            CompletableFuture<Integer> f5 = f2.thenCombine(f3, (result2, result3) -> {
        
                Integer result5 = result2 + result3;
                Long endTIME = System.currentTimeMillis();
                Long elapsedTime = endTIME - startTIME;
                System.out.println("任务五 elapsedTime =" + elapsedTime);
                return result5;
            });
    
            /* 任务六 （OR汇聚）： 任务二 或 任务三 其中一个完成了就开始执行 */
            CompletableFuture<Integer> f6 = f2.applyToEither(f3, result -> {
        
                Integer result6 = result;
                Long endTIME = System.currentTimeMillis();
                Long elapsedTime = endTIME - startTIME;
                System.out.println("任务六 elapsedTime = " + elapsedTime);
                return result6;
            });
    
            System.out.println("任务四 result ：" + f4.join());
            System.out.println("任务五 result ：" + f5.join());
            System.out.println("任务六 result ：" + f6.join());
    
            executorService.shutdown();
        }
    
    }

### 4. 线程池 + `CompletionService` ###

以下使用场景可以使用 CompletionService 来实现：

1.  多个任务并发执行，先完成的任务执行下一步骤
2.  相同任务分给多个线程并发执行，其中一个返回结果就结束剩余的所有任务

public class ExampleFive {
        
    
        public static void main(String[] args) {
        
            System.out.println("【主线程】开始运行");
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            Long startTIME = System.currentTimeMillis();
            // 创建CompletionService
            CompletionService<Integer> completionService = new ExecutorCompletionService<>(executorService);
    
            completionService.submit(()->{
        
                Thread.sleep(3000);
                Long endTIME = System.currentTimeMillis();
                Long elapsedTime = endTIME - startTIME;
                System.out.println("任务一 elapsedTime = " + elapsedTime);
                return 1;
            });
    
            completionService.submit(()->{
        
                Thread.sleep(1000);
                Long endTIME = System.currentTimeMillis();
                Long elapsedTime = endTIME - startTIME;
                System.out.println("任务二 elapsedTime = " + elapsedTime);
                return 2;
            });
    
            completionService.submit(()->{
        
                Thread.sleep(2000);
                Long endTIME = System.currentTimeMillis();
                Long elapsedTime = endTIME - startTIME;
                System.out.println("任务三 elapsedTime = " + elapsedTime);
                return 3;
            });
    
            for(int i =0 ; i<3;i++) {
        
                try{
        
                    Future<Integer> future= completionService.take();
                    Integer r = future.get();
                    System.out.println("任务结果 = " + r);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }
            System.out.println("【主线程】结束运行");
        }
    }