MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java 无界队列在线程池中的影响

2023-10-032.6k 阅读

Java 线程池基础

在深入探讨无界队列在线程池中的影响之前，我们先来回顾一下 Java 线程池的基础知识。

线程池的概念

线程池是一种管理和复用线程的机制。在传统的线程使用方式中，每当有任务需要执行时，就创建一个新的线程，任务执行完毕后，线程销毁。这种方式在高并发场景下会带来巨大的开销，因为线程的创建和销毁是比较昂贵的操作。线程池通过维护一组预先创建好的线程，当有任务到来时，从线程池中取出一个空闲线程来执行任务，任务完成后，线程并不销毁，而是返回线程池等待下一个任务，这样就大大减少了线程创建和销毁的开销。

Java 线程池的实现类

在 Java 中，线程池的主要实现类是 ThreadPoolExecutor。它提供了丰富的构造函数和方法，允许开发者根据不同的需求进行定制化配置。以下是 ThreadPoolExecutor 常用的构造函数：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}
  • corePoolSize：核心线程数，线程池在正常情况下保持的线程数量。即使这些线程处于空闲状态，它们也不会被销毁，除非设置了 allowCoreThreadTimeOut 为 true。
  • maximumPoolSize：最大线程数，线程池允许创建的最大线程数量。当任务队列已满且线程数小于最大线程数时，线程池会创建新的线程来处理任务。
  • keepAliveTime：线程存活时间，当线程数大于核心线程数时，多余的空闲线程在多长时间后会被销毁。
  • unit：keepAliveTime 的时间单位，如 TimeUnit.SECONDS 表示秒。
  • workQueue：任务队列，用于存储等待执行的任务。当线程池中的所有核心线程都在忙碌时，新提交的任务会被放入这个队列中等待执行。
  • threadFactory：线程工厂，用于创建新的线程。通过线程工厂可以定制线程的名称、优先级等属性。
  • handler：拒绝策略，当任务队列已满且线程数达到最大线程数时，新提交的任务会被拒绝，由拒绝策略来决定如何处理这些被拒绝的任务。

线程池的工作流程

  1. 当一个新任务提交到线程池时，首先检查核心线程是否都在忙碌。如果有核心线程空闲，则直接分配该任务给空闲的核心线程执行。
  2. 如果所有核心线程都在忙碌，检查任务队列是否已满。如果任务队列未满，则将任务放入任务队列中等待执行。
  3. 如果任务队列已满，检查当前线程数是否小于最大线程数。如果小于最大线程数，则创建一个新的线程来执行该任务。
  4. 如果当前线程数已经达到最大线程数，那么根据设置的拒绝策略来处理这个新任务。常见的拒绝策略有以下几种：
    • AbortPolicy：默认的拒绝策略，直接抛出 RejectedExecutionException 异常。
    • CallerRunsPolicy：将任务交回给调用者线程来执行。
    • DiscardPolicy：直接丢弃该任务，不做任何处理。
    • DiscardOldestPolicy：丢弃任务队列中最老的任务，然后尝试将新任务放入队列。

队列在 Java 线程池中的作用

任务队列是线程池的重要组成部分，它在调节任务处理节奏和线程负载方面起着关键作用。

任务缓冲

任务队列作为一个缓冲区，当核心线程都在忙碌时，新提交的任务可以暂时存储在队列中。这样可以避免立即创建过多的线程，从而减少系统资源的消耗。例如，在一个 Web 服务器应用中，可能会同时收到大量的 HTTP 请求，这些请求可以被放入任务队列，然后由线程池中的线程依次取出并处理。

控制线程创建

通过合理设置任务队列的容量和线程池的参数，可以有效地控制线程的创建数量。如果任务队列容量较大，那么在任务高峰期，线程池可以通过队列来缓冲任务，而不需要立即创建大量的线程。只有当队列已满且任务数量持续增加时，才会考虑创建新的线程直到达到最大线程数。这有助于防止系统因为创建过多线程而导致资源耗尽。

平滑任务处理

任务队列可以平滑任务的处理过程。假设系统在某个时间段内收到了大量的突发任务，如果没有任务队列，这些任务可能会导致线程池瞬间创建大量线程，造成系统负载过高。而有了任务队列，任务可以按照队列的顺序依次被处理，避免了任务处理的剧烈波动，提高了系统的稳定性。

Java 中的无界队列

在 Java 中，有多种类型的队列可供选择，其中无界队列在某些场景下有着独特的应用。

无界队列的定义

无界队列是指理论上可以无限添加元素的队列。在 Java 中，LinkedBlockingQueue 如果不指定容量，就是一个无界队列。它内部使用链表结构来存储元素，因此在添加元素时，只要内存足够，就不会受到容量的限制。

常用的无界队列实现

  1. LinkedBlockingQueue：如前所述，LinkedBlockingQueue 是一个基于链表的阻塞队列。当不指定容量时，它是无界的。它提供了高效的插入和移除操作，适用于高并发场景。以下是创建一个无界 LinkedBlockingQueue 的示例：
BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
  1. PriorityBlockingQueue：这是一个支持优先级排序的无界阻塞队列。元素按照自然顺序或者自定义的比较器进行排序。在任务队列中，如果任务有优先级之分，PriorityBlockingQueue 就非常有用。例如，系统监控任务可能比普通业务任务具有更高的优先级，就可以使用这个队列来保证高优先级任务优先执行。示例代码如下：
PriorityBlockingQueue<Runnable> priorityQueue = new PriorityBlockingQueue<>(10, (r1, r2) -> {
    // 假设 Runnable 实现了具有优先级的接口
    int priority1 = ((PrioritizedTask) r1).getPriority();
    int priority2 = ((PrioritizedTask) r2).getPriority();
    return Integer.compare(priority1, priority2);
});

这里假设 PrioritizedTask 是一个实现了优先级接口的任务类。

无界队列在线程池中的影响

资源消耗方面的影响

  1. 内存占用
    • 由于无界队列理论上可以无限添加任务，当任务源源不断地提交到线程池且处理速度较慢时，队列会不断增长，从而占用大量的内存。例如，在一个数据采集系统中，如果数据采集速度很快，而数据处理线程的处理能力有限，无界队列可能会积累大量的数据采集任务，导致内存消耗持续上升，最终可能引发内存溢出错误。
    • 下面通过一个简单的代码示例来演示这种情况：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class UnboundedQueueMemoryExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1,  // corePoolSize
                1,  // maximumPoolSize
                0L,
                TimeUnit.MILLISECONDS,
                unboundedQueue);

        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            executor.submit(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

在这个示例中，我们创建了一个只有一个核心线程和一个最大线程的线程池，使用无界队列 LinkedBlockingQueue。然后不断向线程池中提交任务，每个任务睡眠 1 秒模拟处理时间。随着任务的不断提交，队列会不断增长，最终可能导致内存溢出。 2. 线程数量控制

  • 无界队列会影响线程池对线程数量的控制。由于队列不会满，当任务提交速度大于线程处理速度时，线程池不会创建超过核心线程数的线程（除非设置了 allowCoreThreadTimeOut 且核心线程空闲时间超过 keepAliveTime）。这可能导致任务在队列中大量积压，而线程池中的线程却没有充分利用系统资源。例如，在一个 CPU 密集型的计算任务场景中，如果使用无界队列，即使系统有足够的 CPU 资源可以创建更多线程来加快任务处理，线程池也不会创建新线程，因为队列永远不会满，从而降低了系统的整体处理性能。

任务处理方面的影响

  1. 任务响应时间
    • 当任务队列中积累了大量任务时，新提交任务的响应时间会显著增加。因为新任务需要在队列中等待前面的任务依次处理完毕。例如，在一个在线交易系统中，交易请求被提交到线程池处理，如果队列中已经积压了大量的请求，新的交易请求可能需要等待很长时间才能被处理，这会严重影响用户体验。
    • 以下代码演示了任务响应时间的变化：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class UnboundedQueueResponseTimeExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1,  // corePoolSize
                1,  // maximumPoolSize
                0L,
                TimeUnit.MILLISECONDS,
                unboundedQueue);

        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            int finalI = i;
            executor.submit(() -> {
                try {
                    Thread.sleep(10);
                    System.out.println("Task " + finalI + " processed at " + (System.currentTimeMillis() - startTime));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                if (!executor.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

在这个示例中，我们创建了一个单线程的线程池和一个无界队列。向线程池中提交 1000 个任务，每个任务模拟处理时间为 10 毫秒。通过打印任务开始处理的时间与初始时间的差值，可以观察到随着任务在队列中的积压，后面任务的响应时间逐渐增加。 2. 任务饥饿

  • 在使用无界队列时，如果某些任务的优先级较高，但由于队列是按照 FIFO（先进先出）原则处理任务，高优先级任务可能会因为前面大量低优先级任务的积压而长时间得不到执行，这就是任务饥饿现象。例如，在一个系统中，系统维护任务具有较高的优先级，但由于大量普通业务任务积压在无界队列中，系统维护任务可能会被延迟执行，影响系统的稳定性和维护效率。如果使用 PriorityBlockingQueue 作为无界队列，并合理设置任务优先级，可以在一定程度上缓解任务饥饿问题，但如果任务提交速度过快，仍然可能出现高优先级任务等待时间过长的情况。

系统稳定性方面的影响

  1. 内存溢出风险
    • 如前面提到的，无界队列可能导致内存不断增长，最终引发内存溢出错误。当内存溢出发生时，整个应用程序可能会崩溃，影响系统的可用性。特别是在一些对稳定性要求极高的系统，如金融交易系统、航空控制系统等，内存溢出可能会带来严重的后果。在实际开发中，需要密切监控系统的内存使用情况，当发现队列中任务数量持续增长且内存占用不断上升时，要及时采取措施，如优化任务处理逻辑、增加线程数量或者调整队列类型等。
  2. 性能抖动
    • 由于无界队列可能导致任务长时间积压，当积压的任务突然被大量处理时，可能会引起系统性能的抖动。例如，在一个批处理系统中，白天业务量较低时，任务在无界队列中不断积压，到了晚上系统进行集中处理时，大量积压的任务同时被处理，可能会导致系统资源瞬间被大量占用，出现性能抖动，影响其他相关业务的正常运行。为了避免这种情况，可以采用流量控制、定时任务分批处理等策略来平滑任务处理过程，减少性能抖动的发生。

如何合理使用无界队列在线程池中

根据业务场景选择

  1. 低负载且任务处理时间短的场景
    • 在一些低负载且任务处理时间较短的业务场景中，使用无界队列是比较合适的。例如，一个小型的 Web 应用，每天的请求量相对较少，且每个请求的处理时间都在毫秒级。在这种情况下，无界队列可以简单有效地缓冲少量的突发请求，而不会造成内存占用过高或任务处理延迟过大的问题。
    • 以下是一个简单的代码示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class LowLoadUnboundedQueueExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,  // corePoolSize
                2,  // maximumPoolSize
                0L,
                TimeUnit.MILLISECONDS,
                unboundedQueue);

        for (int i = 0; i < 10; i++) {
            int finalI = i;
            executor.submit(() -> {
                System.out.println("Task " + finalI + " processed quickly");
            });
        }
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                if (!executor.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

在这个示例中，模拟了一个低负载场景，创建了一个有两个核心线程和最大线程的线程池，使用无界队列。提交少量任务，每个任务简单打印信息，由于任务处理快且负载低，无界队列能够很好地工作。 2. 对任务顺序严格要求且任务量可预测的场景

  • 如果业务场景对任务的执行顺序有严格要求，且任务量在可预测的范围内，无界队列也能发挥作用。例如，在一个日志记录系统中，需要按照事件发生的顺序记录日志，并且每天产生的日志量是相对稳定的。此时使用无界队列可以保证日志任务按照顺序依次处理，而不用担心队列溢出的问题。

结合其他参数优化

  1. 调整核心线程数和最大线程数
    • 当使用无界队列时，可以适当调整核心线程数和最大线程数来优化系统性能。如果任务处理时间较长，可以增加核心线程数，让更多的任务能够同时处理，减少任务在队列中的积压。例如，在一个视频转码应用中，每个转码任务需要较长时间处理，通过增加核心线程数，可以提高转码效率，同时无界队列可以缓冲少量的突发任务。
    • 以下是调整核心线程数和最大线程数的示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class AdjustThreadNumbersExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,  // corePoolSize
                10,  // maximumPoolSize
                10L,
                TimeUnit.SECONDS,
                unboundedQueue);

        for (int i = 0; i < 50; i++) {
            int finalI = i;
            executor.submit(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("Task " + finalI + " processed");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                if (!executor.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

在这个示例中，我们设置了 5 个核心线程和 10 个最大线程，当任务提交数量较多且处理时间为 1 秒时，适当增加线程数可以加快任务处理，减少队列积压。 2. 设置合理的拒绝策略

  • 虽然无界队列理论上不会满，但在实际应用中，由于系统资源的限制，可能会出现类似于队列满的情况（如内存不足）。此时，设置合理的拒绝策略非常重要。例如，可以设置 CallerRunsPolicy 拒绝策略，当出现异常情况时，将任务交回给调用者线程处理，这样可以避免任务丢失，同时也能让调用者感知到系统的负载情况，采取相应的措施，如降低任务提交频率等。
  • 以下是设置拒绝策略的示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CallerRunsPolicy;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class RejectionPolicyExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1,  // corePoolSize
                1,  // maximumPoolSize
                0L,
                TimeUnit.MILLISECONDS,
                unboundedQueue,
                new CallerRunsPolicy());

        for (int i = 0; i < 1000; i++) {
            int finalI = i;
            executor.submit(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("Task " + finalI + " processed");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                if (!executor.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

在这个示例中，我们设置了 CallerRunsPolicy 拒绝策略，当线程池出现异常情况无法处理任务时，任务会由调用者线程执行，通过打印信息可以观察到任务的执行情况。

监控与调优

  1. 监控队列大小和任务处理情况
    • 在运行时监控无界队列的大小和任务的处理情况是非常必要的。可以通过 ThreadPoolExecutor 提供的方法获取队列的当前大小、已完成任务数等信息。例如，通过定时打印队列大小，可以及时发现队列是否有异常增长的趋势。
    • 以下是监控队列大小的示例代码：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class QueueMonitoringExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> unboundedQueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,  // corePoolSize
                2,  // maximumPoolSize
                0L,
                TimeUnit.MILLISECONDS,
                unboundedQueue);

        for (int i = 0; i < 50; i++) {
            int finalI = i;
            executor.submit(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("Task " + finalI + " processed");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 定时监控队列大小
        Thread monitorThread = new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(5000);
                    System.out.println("Queue size: " + executor.getQueue().size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        monitorThread.setDaemon(true);
        monitorThread.start();

        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                if (!executor.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

在这个示例中，我们启动了一个守护线程来定时打印队列的大小，通过观察队列大小的变化，可以及时发现任务积压等问题。 2. 根据监控结果进行调优

  • 根据监控得到的数据，如队列大小持续增长、任务处理时间过长等，可以对线程池的参数进行调整。如果发现队列大小不断增加且核心线程一直处于忙碌状态，可以考虑增加核心线程数；如果发现任务处理时间过长，可以优化任务的处理逻辑，提高处理效率。通过不断地监控和调优，可以使线程池在使用无界队列时保持最佳的性能状态。

综上所述，无界队列在 Java 线程池中既有其优势，也存在一些潜在的问题。在实际应用中，需要根据具体的业务场景，合理选择和配置线程池及无界队列，并通过监控和调优来确保系统的稳定性和性能。