Java 线程池简介

缺乏、安全感 2022-08-31 07:24 143阅读 0赞

Java 线程池简介

### 文章目录 ###

*  1.介绍
 *  2.线程池
 *  3.Java中的线程池
 *   *  3.1. Executors, Executor,ExecutorService
     *   *  3.1.1. Executor
         *  3.1.2. ExecutorService
     *  3.2. ThreadPoolExecutor
     *   *  3.2.1. newFixedThreadPool
         *  3.2.2.Executors.newCachedThreadPool()
         *  3.2.3.Executors.newSingleThreadExecutor()
     *  3.3.ScheduledThreadPoolExecutor
     *  3.4. ForkJoinPool

# 1.介绍 #

介绍 Java 中的线程池

# 2.线程池 #

在 Java 中，线程被映射到系统级线程，这是操作系统的资源。 如果不受控制地创建线程，可能很快就会耗尽这些资源。

操作系统也在线程之间进行上下文切换——以模拟并行性。 一个简单的观点是，我们产生的线程越多，每个线程花在实际工作上的时间就越少。

线程池模式有助于在多线程应用程序中节省资源并将并行性包含在某些预定义的限制中。

当使用线程池时，以并行任务的形式编写并发代码，并将它们提交给线程池的一个实例执行。 该实例控制多个重复使用的线程来执行这些任务。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70_pic_center]

该模式允许控制应用程序创建的线程数量及其生命周期。 还能够安排任务的执行并将传入的任务保持在队列中。

# 3.Java中的线程池 #

## 3.1. Executors, Executor,ExecutorService ##

Executors 帮助器类包含多个用于创建预配置线程池实例的方法。 如果不需要任何自定义，可以使用它们。

使用 Executor 和 ExecutorService 接口来处理 Java 中不同的线程池实现。 通常，应该让我们的代码与线程池实现解耦，并在整个应用程序中使用这些接口。

### 3.1.1. Executor ###

Executor 接口有一个 execute 方法来提交 Runnable 实例以供执行。

看一个示例，了解如何使用 Executors API 获取由单个线程池和无界队列支持的 Executor 实例，用于顺序执行任务。

Executor executor = Executors.newSingleThreadExecutor();
    executor.execute(() -> System.out.println("Hello World"));

### 3.1.2. ExecutorService ###

ExecutorService 接口包含大量方法来控制任务的进度和管理服务的终止。 使用这个接口，可以提交任务执行，还可以使用返回的 Future 实例控制它们的执行。

以下示例将创建一个 ExecutorService，提交一个任务，然后使用返回的 Future 的 get 方法等待提交的任务完成并返回值：

ExecutorService executorService = Executors.newFixedThreadPool(10);
    Future<String> future = executorService.submit(() -> "Hello World");
    // 其他操作
    String result = future.get();

Runnable 的单一方法不会抛出异常，也不会返回值。 Callable 接口可能更方便，因为它允许我们抛出异常并返回一个值。

最后，为了让编译器推断 Callable 类型，只需从 lambda 返回一个值。

## 3.2. ThreadPoolExecutor ##

ThreadPoolExecutor 是一个可扩展的线程池实现，具有许多参数和钩子。

我们将在这里讨论的主要配置参数是 corePoolSize、maximumPoolSize 和 keepAliveTime。

该池由始终保持在内部的固定数量的核心线程组成。 它还包含一些多余的线程，这些线程可能会在不再需要时产生并终止。

corePoolSize 参数是将被实例化并保留在池中的核心线程数。 当有新任务进来时，**如果所有核心线程都忙且内部队列已满，则允许池增长到maximumPoolSize**。

keepAliveTime 参数是允许过多线程（实例化超过 corePoolSize）处于空闲状态的时间间隔。 默认情况下，ThreadPoolExecutor 只考虑非核心线程进行删除。 为了对核心线程应用相同的移除策略，我们可以使用 allowCoreThreadTimeOut(true) 方法。

这些参数涵盖了广泛的用例，但最典型的配置是在 Executors 静态方法中预定义的。

### 3.2.1. newFixedThreadPool ###

newFixedThreadPool 方法创建一个 ThreadPoolExecutor，其 corePoolSize 和 maximumPoolSize 参数值相等，且 keepAliveTime 为零。 这意味着此线程池中的线程数始终相同：

public static void main(String[] args) { 
            ThreadPoolExecutor executor =
                    (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
            executor.submit(() -> { 
                Thread.sleep(1000);
                return null;
            });
            executor.submit(() -> { 
                Thread.sleep(1000);
                return null;
            });
            executor.submit(() -> { 
                Thread.sleep(1000);
                return null;
            });
    
            System.out.println(executor.getPoolSize());//2
            System.out.println(executor.getQueue().size());//1
            executor.shutdown();
        }

在这里，实例化了一个固定线程数为 2 的 ThreadPoolExecutor。这意味着如果同时运行的任务数总是小于或等于 2，它们就会立即执行。 否则，其中一些任务可能会被放入队列以等待轮到他们。

创建了三个 Callable 任务，它们通过休眠 1000 毫秒来模拟正在运行中。 前两个任务将立即运行，第三个任务将不得不在队列中等待。 可以通过在提交任务后立即调用 getPoolSize() 和 getQueue().size() 方法来验证它。

### 3.2.2.Executors.newCachedThreadPool() ###

可以使用 Executors.newCachedThreadPool() 方法创建另一个预配置的 ThreadPoolExecutor。 此方法根本不接收多个线程。 将 corePoolSize 设置为 0，并将 maximumPoolSize 设置为 Integer.MAX\_VALUE。 最后，keepAliveTime 为 60 秒：

public static void main(String[] args) { 
            ThreadPoolExecutor executor =
                    (ThreadPoolExecutor) Executors.newCachedThreadPool();
            executor.submit(() -> { 
                Thread.sleep(1000);
                return null;
            });
            executor.submit(() -> { 
                Thread.sleep(1000);
                return null;
            });
            executor.submit(() -> { 
                Thread.sleep(1000);
                return null;
            });
    
            System.out.println(executor.getPoolSize());//3
            System.out.println(executor.getQueue().size());//1
            executor.shutdown();
        }

这些参数值意味着缓存的线程池可以无限增长以容纳任意数量的提交任务。 但是当不再需要线程时，它们将在 60 秒不活动后被处理掉。 一个典型的用例是当我们的应用程序中有很多短期任务时。

队列大小将始终为零，因为在内部使用 SynchronousQueue 实例。 在 SynchronousQueue 中，插入和删除操作总是同时发生。 因此，队列实际上从未包含任何内容。

### 3.2.3.Executors.newSingleThreadExecutor() ###

Executors.newSingleThreadExecutor() API 创建了另一种典型的 ThreadPoolExecutor 形式，其中包含单个线程。 单线程执行器是创建事件循环的理想选择。 corePoolSize 和 maximumPoolSize 参数等于 1，keepAliveTime 为 0。Executors.newSingleThreadExecutor() API 创建了另一种典型形式的 ThreadPoolExecutor 包含单个线程。 单线程执行器是创建事件循环的理想选择。 corePoolSize 和 maximumPoolSize 参数等于 1，keepAliveTime 为 0。

public static void main(String[] args) throws InterruptedException { 
            AtomicInteger counter = new AtomicInteger();
            CountDownLatch countDownLatch = new CountDownLatch(3);
    
            ExecutorService executor = Executors.newSingleThreadExecutor();
            executor.submit(() -> { 
                counter.set(1);
                countDownLatch.countDown();
            });
            executor.submit(() -> { 
                counter.compareAndSet(1, 2);
                countDownLatch.countDown();
            });
            executor.submit(() -> { 
                counter.compareAndSet(2, 3);
                countDownLatch.countDown();
            });
            countDownLatch.await();
            System.out.println(counter.get());//3
            executor.shutdown();
        }

上例中的任务将按顺序运行，因此任务完成后标志值为 3：

另外，这个 ThreadPoolExecutor 用一个不可变的包装器装饰，所以它不能在创建后重新配置。 请注意，这也是无法将其转换为 ThreadPoolExecutor 的原因。

## 3.3.ScheduledThreadPoolExecutor ##

ScheduledThreadPoolExecutor 扩展了 ThreadPoolExecutor 类，并且还通过几个附加方法实现了 ScheduledExecutorService 接口：

*  schedule 方法允许在指定的延迟后运行一次任务。
 *  scheduleAtFixedRate 方法允许在指定的初始延迟后运行任务，然后在特定时间段内重复运行它。 period 参数是任务开始时间之间测量的时间，因此执行率是固定的。
 *  scheduleWithFixedDelay 方法与 scheduleAtFixedRate 类似，因为它重复运行给定的任务，但指定的延迟是在前一个任务结束和下一个任务开始之间测量的。 执行率可能因运行任何给定任务所需的时间而异

我们通常使用 Executors.newScheduledThreadPool() 方法创建一个 ScheduledThreadPoolExecutor，它具有给定的 corePoolSize、无界的 maximumPoolSize 和零 keepAliveTime。

以下是如何安排任务在 1000 毫秒内执行：

public static void main(String[] args) { 
            ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
            executor.schedule(() -> { 
                System.out.println("Hello World");
            }, 1000, TimeUnit.MILLISECONDS);
    
            executor.shutdown();
        }

以下代码显示了如何在 500 毫秒延迟后运行任务，然后每 100 毫秒重复一次。 调度任务后，等待它使用 CountDownLatch 锁触发 3 次。 然后我们使用 Future.cancel() 方法取消它：

CountDownLatch lock = new CountDownLatch(3);
            ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
            ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> { 
                System.out.println("Hello World");
                lock.countDown();
            }, 500, 100, TimeUnit.MILLISECONDS);
    
            lock.await(1000, TimeUnit.MILLISECONDS);
            future.cancel(true);

## 3.4. ForkJoinPool ##

ForkJoinPool 是 Java 7 中引入的 fork/join 框架的核心部分。它解决了递归算法中产生多个任务的常见问题。 使用简单的 ThreadPoolExecutor 很快会耗尽线程，因为每个任务或子任务都需要运行自己的线程。

在 fork/join 框架中，任何任务都可以产生（fork）多个子任务并使用 join 方法等待它们完成。 fork/join 框架的好处是它不会为每个任务或子任务创建一个新线程，而是实现了工作窃取算法。

看一个使用 ForkJoinPool 遍历节点树并计算所有叶值总和的简单示例。 下面是一个由一个节点、一个 int 值和一组子节点组成的树的简单实现：

public class TreeNode { 
        int value;
    
        Set<TreeNode> children;
    
        TreeNode(int value, TreeNode... children) { 
            this.value = value;
            this.children = Sets.newHashSet(children);
        }
    }

现在，如果想并行地对树中的所有值求和，需要实现一个 RecursiveTask 接口。 每个任务接收自己的节点并将其值添加到其子节点的值之和中。 为了计算子值的总和，任务实现执行以下操作：

*  设置孩子节点的流
 *  映射到这个流，为每个元素创建一个新的 CountingTask
 *  通过分叉来运行每个子任务
 *  通过在每个分叉任务上调用 join 方法来收集结果
 *  使用 Collectors.summingInt 收集器对结果求和

public class CountingTask extends RecursiveTask<Integer> { 
    
        private final TreeNode node;
    
        public CountingTask(TreeNode node) { 
            this.node = node;
        }
    
        @Override
        protected Integer compute() { 
                return node.value + node.children.stream()
                    .map(childNode -> new CountingTask(childNode).fork()) //.fork()将任务加入工作队列中
                    .collect(Collectors.summingInt(ForkJoinTask::join));//等待任务执行完毕
        }
    }

运行

public static void main(String[] args) { 
            TreeNode tree = new TreeNode(5,
                    new TreeNode(3), new TreeNode(2,
                    new TreeNode(2), new TreeNode(8)));
    
            ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
            int sum = forkJoinPool.invoke(new CountingTask(tree));
            System.out.println(sum);
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70_pic_center]: /images/20220829/5fd373dda00240668dbe0fdcc0ce42f7.png