多线程 - Executor

╰+攻爆jí腚メ 2024-03-22 14:12 27阅读 0赞

## 使用Executor接口来实现线程池 ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     *
     * Executor是Java中用于实现线程池的一个接口，它定义了线程池应该具有的基本方法和属性。
     * 使用线程池可以优化多线程程序，减少线程创建和销毁的开销，提高线程的复用率和性能。
     *
     *
     * 使用Executor接口来实现线程池。
     * 首先通过Executors类的静态方法newFixedThreadPool创建一个固定大小的线程池，其中包含5个线程。
     * 然后，提交了10个任务给线程池执行，每个任务都是一个MyRunnable对象。
     * 每个MyRunnable对象都实现了Runnable接口，并包含了一个processMessage方法，该方法模拟了一个长时间的任务。
     * 最后，我们使用shutdown方法关闭线程池，并等待所有任务执行完毕。
     *
     * 在输出中，可以看到5个线程依次处理了10个任务，并在处理完每个任务后释放线程。
     * 由于线程池的大小是5，因此每次只会有5个任务同时执行，其余任务会在队列中等待。
     * 这样可以避免线程过多导致系统负载过大的问题，同时也提高了程序的性能和稳定性。
     */
    public class MyExecutor {
        
        public static void main(String[] args) {
        
            // 创建线程池，其中包含5个线程
            ExecutorService executor = Executors.newFixedThreadPool(5);
    
            // 提交10个任务给线程池执行
            for (int i = 0; i < 10; i++) {
        
                Runnable worker = new MyRunnable("" + i);
                executor.execute(worker);
            }
    
            // 关闭线程池
            executor.shutdown();
            while (!executor.isTerminated()) {
        
            }
    
            System.out.println("All tasks have been completed.");
        }
    }
    
    class MyRunnable implements Runnable {
        
        private String message;
    
        public MyRunnable(String message) {
        
            this.message = message;
        }
    
        @Override
        public void run() {
        
            System.out.println(Thread.currentThread().getName() + " Start. Message = " + message);
            processMessage();
            System.out.println(Thread.currentThread().getName() + " End.");
        }
    
        private void processMessage() {
        
            try {
        
                Thread.sleep(2000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

## ExecutorService ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.Future;
    
    /**
     * ExecutorService是Executor接口的一个扩展，它提供了更加丰富的线程池管理功能，
     * 例如线程池大小的动态调整、任务提交的结果返回、任务取消等。可以说，ExecutorService是一个比Executor更加高级和实用的线程池管理接口。
     *
     * 和前面介绍的Executor示例非常相似，只是使用了ExecutorService的submit方法来提交任务，并返回Future对象。
     * Future对象可以用来检查任务的执行状态，例如是否已经完成、是否抛出异常等。
     * 另外，我们使用了ExecutorService的awaitTermination方法来等待所有任务执行完毕，该方法会阻塞当前线程，直到所有任务执行完毕或者超时。
     *
     * 需要注意的是，使用ExecutorService提交的任务是异步执行的，即提交任务后就立即返回，并不会等待任务执行完成。
     * 如果需要等待任务完成并获取结果，可以使用Future对象来获取任务执行的状态和结果。
     * 同时，需要在程序结束前调用ExecutorService的shutdown方法来关闭线程池，释放资源。
     */
    public class MyExecutorService {
        
        public static void main(String[] args) {
        
            // 创建线程池，其中包含5个线程
            ExecutorService executor = Executors.newFixedThreadPool(5);
    
            // 提交10个任务给线程池执行，并且返回Future对象
            for (int i = 0; i < 10; i++) {
        
                Runnable worker = new MyRunnable("" + i);
                Future<?> future = executor.submit(worker);
            }
    
            // 关闭线程池，等待所有任务执行完毕
            executor.shutdown();
            try {
        
                executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
    
            System.out.println("All tasks have been completed.");
        }
    }

## ExecutorCompletionService ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.*;
    
    /**
     *
     * ExecutorCompletionService是CompletionService的一个实现类，它利用了Executor框架，通过将任务提交给一个线程池来异步执行任务。
     *
     * 和CompletionService一样，ExecutorCompletionService也可以将异步执行的任务结果按照完成顺序存储在BlockingQueue中，
     * 可以方便地获取最先完成的任务结果。ExecutorCompletionService的实现方式是在CompletionService的基础上，
     * 封装了一个内部类QueueingFuture，它继承了FutureTask，并重写了done()方法，将已完成的任务的Future对象存储在BlockingQueue中。
     *
     */
    public class MyExecutorCompletionService {
        
        public static void main(String[] args) throws InterruptedException, ExecutionException {
        
            ExecutorService executor = Executors.newFixedThreadPool(10);
            ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);
    
            for (int i = 0; i < 10; i++) {
        
                completionService.submit(() -> {
        
                    int sleepTime = (int) (Math.random() * 1000);
                    Thread.sleep(sleepTime);
                    return sleepTime;
                });
            }
    
            for (int i = 0; i < 10; i++) {
        
                Future<Integer> future = completionService.take();
                System.out.println(future.get());
            }
    
            executor.shutdown();
        }
    }

## Future ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.*;
    
    /**
     * Future接口通常和ExecutorService一起使用，ExecutorService的submit()方法可以返回一个Future对象，表示提交的任务的执行结果。
     *
     *
     * 使用ExecutorService的submit()方法提交一个Callable对象，表示一个异步计算。
     * submit()方法返回一个Future对象，表示异步计算的结果。可以调用Future的get()方法来获取计算的结果，
     * 如果计算还没有完成，get()方法将阻塞直到计算完成。最后，调用ExecutorService的shutdown()方法关闭线程池。
     */
    public class MyFuture {
        
        public static void main(String[] args) {
        
            ExecutorService executor = Executors.newFixedThreadPool(1);
    
            Future<Integer> future = executor.submit(() -> {
        
                Thread.sleep(2000);
                return 2;
            });
    
            try {
        
                System.out.println("Waiting for result...");
                Integer result = future.get();
                System.out.println("Result is " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            } finally {
        
                executor.shutdown();
            }
        }
    }

## FutureTask ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.*;
    
    /**
     * FutureTask是Java提供的一个类，它实现了Future接口和Runnable接口，可以作为Runnable对象被线程执行，也可以作为Future对象获取异步计算的结果。
     *
     * FutureTask有三个构造方法：
     *
     * FutureTask(Callable<V> callable)：接收一个Callable对象，表示一个异步计算。
     *
     * FutureTask(Runnable runnable, V result)：接收一个Runnable对象和一个结果对象，表示一个异步计算，当异步计算完成时，它将返回给定的结果对象。
     *
     * FutureTask(Runnable runnable)：接收一个Runnable对象，表示一个异步计算，当异步计算完成时，它将返回null。
     *
     *
     * 使用ExecutorService的submit()方法提交一个FutureTask对象，表示一个异步计算。
     * 可以调用FutureTask的get()方法来获取计算的结果，如果计算还没有完成，get()方法将阻塞直到计算完成。
     * 最后，调用ExecutorService的shutdown()方法关闭线程池。
     *
     * 需要注意的是，FutureTask对象只能被执行一次，如果试图对同一个FutureTask对象多次调用run()方法或提交到线程池中执行，
     * 将会抛出IllegalStateException异常。
     *
     */
    public class MyFutureTask {
        
        public static void main(String[] args) {
        
            ExecutorService executor = Executors.newFixedThreadPool(1);
    
            FutureTask<Integer> futureTask = new FutureTask<>(() -> {
        
                Thread.sleep(2000);
                return 2;
            });
    
            executor.submit(futureTask);
    
            try {
        
                System.out.println("Waiting for result...");
                Integer result = futureTask.get();
                System.out.println("Result is " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            } finally {
        
                executor.shutdown();
            }
        }
    }

## ThreadPoolExecutor ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.*;
    
    /**
     * ThreadPoolExecutor是Java提供的一个灵活的线程池实现，它是ExecutorService接口的实现类，提供了更加丰富的线程池管理功能。
     * ThreadPoolExecutor可以通过自定义配置线程池参数来满足不同场景下的需求，例如线程池大小、任务队列、拒绝策略等。
     *
     *
     * 创建了一个ThreadPoolExecutor对象，并使用自定义的配置参数来创建线程池。
     * 其中，corePoolSize指定了线程池中的核心线程数为5，maxPoolSize指定了线程池中最大线程数为10，
     * keepAliveTime指定了线程空闲时间为5秒，queue指定了任务队列为一个无界的LinkedBlockingQueue。
     *
     * 接着，提交10个任务给线程池执行，任务会先分配给核心线程执行，如果核心线程都在执行任务，则将任务加入任务队列，
     * 如果任务队列已满，则创建新的线程执行任务，直到线程池中的线程数达到最大值。
     * 当线程空闲时间超过keepAliveTime时，空闲线程会被终止并移出线程池。
     *
     * 需要注意的是，使用ThreadPoolExecutor需要根据具体场景进行合理的配置，
     * 例如根据任务性质、任务量、系统资源等来确定线程池大小、任务队列大小等参数，
     * 否则可能会导致线程池性能不佳、任务无法及时响应等问题。
     */
    public class MyThreadPoolExecutor {
        
        public static void main(String[] args) {
        
            // 创建一个线程池，其中包含5个线程
            int corePoolSize = 5;
            int maxPoolSize = 10;
            long keepAliveTime = 5000;
            BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
            ExecutorService executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime,
                    TimeUnit.MILLISECONDS, queue);
    
            // 提交10个任务给线程池执行
            for (int i = 0; i < 10; i++) {
        
                Runnable worker = new MyRunnable("" + i);
                executor.execute(worker);
            }
    
            // 关闭线程池，等待所有任务执行完毕
            executor.shutdown();
            try {
        
                executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
    
            System.out.println("All tasks have been completed.");
        }
    }

## CompletionService ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.*;
    
    /**
     * CompletionService是Java中用于异步执行任务并按照完成顺序获取结果的一个工具类。
     * 它提供了一个Executor和一个BlockingQueue，可以将异步执行的任务结果按照完成顺序存储在BlockingQueue中，
     * 可以方便地获取最先完成的任务结果。
     *
     * CompletionService可以通过submit()方法向Executor提交一个任务，返回一个Future对象，可以通过Future对象获取任务的结果。
     * 它还提供了take()方法，用于从BlockingQueue中获取已完成的任务的Future对象，如果没有已完成的任务，则该方法会阻塞直到有任务完成。
     * 通过CompletionService的take()方法可以方便地获取已完成的任务的结果，而不需要使用Future对象的get()方法等待异步任务的完成。
     *
     *
     *
     * 创建了一个ExecutorService和一个CompletionService对象，使用ExecutorService来执行异步任务，
     * 使用CompletionService来获取已完成的任务的结果。
     * 使用循环提交了10个任务到CompletionService中，每个任务会随机休眠一段时间并返回休眠时间。
     * 接着，使用循环从CompletionService中获取已完成的任务的Future对象，使用Future对象的get()方法获取任务结果并输出。
     * 由于CompletionService会按照完成顺序将已完成的任务的Future对象存储在BlockingQueue中，因此可以方便地获取最先完成的任务结果。
     *
     * 需要注意的是，CompletionService也支持超时等功能，可以根据实际需求进行灵活配置。
     */
    public class MyCompletionService {
        
        public static void main(String[] args) throws InterruptedException, ExecutionException {
        
            ExecutorService executor = Executors.newFixedThreadPool(10);
            CompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);
    
            for (int i = 0; i < 10; i++) {
        
                completionService.submit(() -> {
        
                    int sleepTime = (int) (Math.random() * 1000);
                    Thread.sleep(sleepTime);
                    return sleepTime;
                });
            }
    
            for (int i = 0; i < 10; i++) {
        
                Future<Integer> future = completionService.take();
                System.out.println(future.get());
            }
    
            executor.shutdown();
        }
    }

## CompletableFuture ##

package com.lfsun.highconcurrency000.executor;
    
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.Collectors;
    import java.util.stream.Stream;
    
    /**
     *
     * CompletableFuture是Java 8中引入的一个类，用于处理异步任务，它可以通过回调函数或组合多个CompletableFuture对象来完成复杂的异步操作。
     *
     * CompletableFuture实现了Future和CompletionStage接口，它的主要特点是可以通过链式调用方法来组合异步操作。
     * 它提供了一些静态方法，例如supplyAsync()、runAsync()等，用于异步执行一个任务并返回CompletableFuture对象。它也提供了一些方法，
     * 例如thenApply()、thenAccept()、thenRun()、thenCompose()等，用于对已完成的任务进行处理，返回一个新的CompletableFuture对象。
     *
     * 创建了三个CompletableFuture对象来模拟三个异步操作，分别是future1、future2、future3。
     * 使用CompletableFuture的allOf()方法将三个CompletableFuture对象聚合起来，返回一个CompletableFuture对象，
     * 表示三个异步操作都完成的情况。
     * 接着，使用CompletableFuture的thenApply()方法对三个CompletableFuture对象的结果进行处理，
     * 将它们聚合到一个Stream对象中。
     * 最后，使用get()方法等待异步操作完成，并遍历Stream对象，输出结果。
     *
     * 需要注意的是，CompletableFuture支持链式调用，也支持异常处理、超时等功能，可以根据实际需求进行灵活配置。
     */
    public class MyCompletableFutureExample {
        
        public static void main(String[] args) throws InterruptedException, ExecutionException {
        
            CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        
                try {
        
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
        
                    throw new IllegalStateException(e);
                }
                return "Result of Future 1";
            });
    
            CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        
                try {
        
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
        
                    throw new IllegalStateException(e);
                }
                return "Result of Future 2";
            });
    
            CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
        
                try {
        
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
        
                    throw new IllegalStateException(e);
                }
                return "Result of Future 3";
            });
    
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2, future3);
    
            CompletableFuture<Stream<String>> allResults = allFutures.thenApply(v -> Stream.of(future1, future2, future3)
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList())
                    .stream());
    
            allResults.get().forEach(System.out::println);
        }
    }