深度思考线程池面经

推荐

1 线程池详解（ThreadPoolExecutor）
2 面试必问的线程池，你懂了吗？
3 Java 中的线程池是怎样实现的？
4 高频面试题：Java线程池应该如何使用？

1 系统设计类题目

1.1 实现一个线程池，效果是这样的：当流量大的时候，线程数要多一点，当流量小的时候，缩的小一些，如果说线程数过多，线程池又不要炸掉，然后客户端可以阻塞

1 首先需要一个有界队列，同时设置一个corePoolSize和一个maxPoolSize参数，分别表示核心线程数以及支持的最大线程数，corePoolSize设置为OS支持的核心线程数+1，根据内存允许的情况设置maxPoolSize参数；另外会为每一个非核心线程设置一个超时时间，当执行完任务后的一段时间没有被分配新任务就自我注销；

2 主线程一般负责将任务分发给线程池中的线程或者添加任务到阻塞队列，同时我还给主线程一个任务那就是每隔一段时间监测当前的流量情况，如果当前的maxPoolSize设置的比较大，但是发现流量在明显降低，则每次都会减少maxPoolSize的值；如果发现当前的流量比较小，但是流量越来越大，则会每次加大maxPoolSize，直到maxPoolSize达到一个系统内存的临界值，这里的检测机制是不断判断当前的等待队列是否已满，并且总线程数是否快要到达maxPoolSize，如果都是则扩容；如果等待队列都没满甚至为空，则可以慢慢减少maxPoolSize。

3 当maxPoolSize加大到系统的临界值并且阻塞队列已经满时，不会执行拒绝策略，主线程不再分配任务给本地的线程池，而是先放入到消息队列中，等待本地的队列中的任务执行了大部分时才将其从消息队列中预取一部分到本地队列。

1.1.1 gpt4对这个方案的评估

您的方案基本上涵盖了线程池的动态调整、任务缓冲以及过载保护的一些核心思路，是一个不错的起点。以下是对您方案的一些补充和建议：

1. 核心线程数设置：

   • 将corePoolSize设置为OS支持的核心线程数+1是一个合理的默认选择。但请记住，核心线程数的最佳设置可能会因应用场景而异。具体的最佳设置可能需要基于负载测试来确定。

2. 动态调整maxPoolSize：

   • 动态地调整maxPoolSize是一个有趣的想法。但请注意，Java的ThreadPoolExecutor默认并不支持动态调整maxPoolSize。您可能需要自定义线程池来实现这个功能。
   • 对于检测机制，您可能需要保证它的粒度不要太细，以避免频繁的调整，这样可能会影响性能。

3. 使用消息队列：

   • 当本地队列满时，将任务放入消息队列是一个不错的备选方案，尤其在微服务环境中。这也为跨节点负载均衡提供了一种机制。
   • 请注意，这种方式会引入额外的延迟和系统复杂性。您需要确保消息队列的可靠性和低延迟。

4. 客户端阻塞：

   • 您提到"客户端可以阻塞"，这意味着当线程池和队列都满时，您想让客户端等待，直到有资源可用。这可以通过使用CyclicBarrier或Semaphore来实现，但要小心，因为这可能会导致客户端超时或其他潜在问题。

5. 线程池饱和策略：

   • 考虑实现自定义的拒绝策略。例如，您可以将任务放入备份存储，如消息队列或数据库，然后在资源可用时再次尝试。

6. 监控与告警：

   • 考虑添加监控和告警功能，当线程池接近饱和或达到某些阈值时，可以发送通知。这有助于系统管理员及时介入并采取必要的措施。

总之，您的方案是一个很好的开始，但要确保在实际环境中进行充分测试和验证。

1.1.3 完善后的版本

主要是动态改变maxPoolSize的粒度上的区别，其他的几乎没变：

1 首先需要一个有界队列，同时设置一个corePoolSize和一个maxPoolSize参数，分别表示核心线程数以及支持的最大线程数，corePoolSize设置为OS支持的核心线程数+1，根据内存允许的情况设置maxPoolSize参数；另外会为每一个非核心线程设置一个超时时间，当执行完任务后的一段时间没有被分配新任务就自我注销；

2 主线程一般负责将任务分发给线程池中的线程或者添加任务到阻塞队列，同时我还给主线程一个任务那就是每隔一段时间监测当前的流量情况，如果当前的maxPoolSize设置的比较大，但是发现流量在明显降低，则粗粒度的减少maxPoolSize的值；如果发现当前的流量比较小，但是流量越来越大，则会一次性粗粒度的加大maxPoolSize，直到maxPoolSize达到一个系统内存的临界值，这里的检测机制是不断判断当前的等待队列是否已满，并且总线程数是否快要到达maxPoolSize，如果都是则扩容；如果等待队列都没满甚至为空，则可以慢慢减少maxPoolSize。

3 当maxPoolSize加大到系统的临界值并且阻塞队列已经满时，不会执行拒绝策略，主线程不再分配任务给本地的线程池，而是先放入到消息队列中，等待本地的队列中的任务执行了大部分时才将其从消息队列中预取一部分到本地队列。

1.1.4 方案二：使用callerRunsPolicy策略以满足题目要求的“客户端可以阻塞”场景

CallerRunsPolicy：调用者运行策略。该策略实现了一种调节机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者（调用线程池执行任务的主线程），从而降低新任务的流程。它不会在线程池的某个线程中执行新提交的任务，而是在一个调用了execute的线程中执行该任务。当线程池的所有线程都被占用，并且工作队列被填满后，下一个任务会在调用execute时在主线程中执行（调用线程池执行任务的主线程）。由于执行任务需要一定时间，因此主线程至少在一段时间内不能提交任务，从而使得工作者线程有时间来处理完正在执行的任务。在这期间，主线程不会调用accept，因此到达的请求将被保存在TCP层的队列中。如果持续过载，那么TCP层将最终发现它的请求队列被填满，因此同样会开始抛弃请求。当服务器过载后，这种过载情况会逐渐向外蔓延开来——从线程池到工作队列到应用程序再到TCP层，最终达到客户端，导致服务器在高负载下实现一种平缓的性能降低。

1.1.5 方案三：基于方案一的动态调整线程池的方法，满足客户端阻塞的另一种方案是，不采用任何拒绝策略，但是当队列满且最大线程都接近阈值时，接收端向发送端发送一个信号，让它慢点发，然后发送端或者客户端就会降低发送速率

1.1.6 方案四：基于方案一的动态调整线程池的方法，满足客户端阻塞的另一种方案是，可以使用发送窗口大小机制去控制发送端的流量，比如在发送端的应用层协议上定义一个发送窗口大小字段，这个字段等于可用的队列节点数加可创建的非核心线程数，当接收端受到响应后，发送的请求数或者任务数不超过这个字段值

1.1.7 客户端本身可以使用同步调用的方式，直到服务端返回响应，客户端才发送下一个请求给这个服务端，这也是一种客户端阻塞。

1.2 面对这里的大流量情况，对于本地内存放不下的任务是不是可以放入到远程服务器的队列中，等待有空闲线程时再去取，但是对于消费者不返回失败，这样设计可以吗？

当面对大流量，本地内存不足以存放待处理的任务时，使用远程服务器的队列（如分布式队列如Kafka, RabbitMQ等）确实是一个可行的策略。这样的设计可以有以下优势：

• 缓解流量压力：即使本地线程池和任务队列已满，也能保证任务不会被直接丢弃。
• 高可用与伸缩性：分布式队列一般设计为高可用，能够水平扩展，以满足大量的写入请求。
• 任务持久化：即使应用或服务器崩溃，已存入远程队列的任务不会丢失。

然而，这种方法也有挑战：

• 网络延迟：与远程队列的通信将引入额外的网络延迟。
• 复杂性增加：引入远程队列需要更复杂的部署和维护策略。
• 成本：增加额外的服务器和网络带宽成本。
• 对于消费者不返回失败，这种设计方式可以增强用户体验和系统的健壮性，但你需要确保系统可以在后续有效地处理这些任务，避免长时间的延迟或其他问题。

1.3 接1.2，一般一个服务通常是集群部署，线程中的队列已满，并且线程数已达到阈值，这个时候我将收到的新任务发给其他的微服务实例可以吗？

当一个服务（微服务）实例的线程池队列已满，并且线程数达到上限时，将新任务转发给同类型的其他微服务实例是一种可能的策略。这样的策略也被称为任务或请求重定向。不过这种策略在实践中应当谨慎使用，并确保考虑以下几点：

1. 负载均衡策略：负载均衡器（例如Nginx, HAProxy, AWS ELB等）通常已经实现了请求分发的策略。在应用层再次实现重定向可能导致某些问题。例如，如果所有实例都转发请求，可能导致循环重定向的问题。

2. 网络开销：请求重定向将导致额外的网络延迟。这可能会影响系统的总体性能，尤其是在高请求率的场景中。

3. 同一集群的压力：如果所有实例都面临类似的压力，即使重定向请求到另一个实例，那么问题仍然可能存在。所以在使用这种策略之前，需要确保其他实例确实有足够的资源来处理重定向的请求。

4. 错误和故障传播：过多地重定向请求可能导致错误或故障的快速传播，从而影响整个服务集群的健康状态。

5. 跟踪和监控：为确保此策略的有效性，需要对转发的请求进行跟踪和监控，以及设置合适的告警。如果重定向的请求太多，可能表明系统存在其他问题，如资源瓶颈、配置问题等。

6. 备选策略：考虑其他备选方案，如扩容实例、优化线程池策略、使用回压机制、异步处理等。

总之，尽管任务或请求重定向是一种可行的策略，但实现和维护它可能相对复杂，并可能引入新的风险。在使用之前，务必进行深入的评估和测试。

1.4 接1.3，一般一个服务通常是集群部署，线程中的队列已满，并且线程数已达到阈值，因为一般客户端发送请求会采取负载均衡策略，所以可能这个集群中的实例都处于忙碌状态，这个时候面新来的请求，我做出告警，让管理者迅速部署或者自动部署新的服务实例，然后服务提供者执行特殊的拒绝策略，告知消费者重定向到刚刚新增的服务实例，这样可以吗？

这种策略涉及到多个层面的操作和通讯。在某些情境中，确实可以考虑这种策略。以下是对这种策略的评估：

1. 自动扩容：许多现代云平台（如AWS、Google Cloud、Azure等）都支持基于特定指标（如CPU利用率、网络I/O、队列长度等）的自动扩容。如果您已经在这些平台上部署了服务，那么您可以配置自动扩容来迅速响应流量增长。

2. 拒绝策略：服务提供者可以实施特殊的拒绝策略，这确实是一种可能的解决方案。但这种策略的效果取决于客户端如何处理这种重定向通知。此外，您需要保证新的服务实例已经准备好并且可以接受这些重定向的请求。

3. 服务发现和注册：为了实现动态重定向到新的服务实例，您需要一个动态的服务发现和注册机制，如Consul、Eureka或其他服务注册中心。这样，新部署的服务实例可以被快速发现并加入到负载均衡器中。

4. 告警：确实，当系统资源达到临界值时，发送告警是很有必要的。这既可以提醒系统管理员采取措施，也可以触发自动响应机制。

5. 网络延迟和额外的开销：请注意，重定向请求会引入额外的网络延迟，而且需要客户端执行额外的操作。这可能会对系统的整体响应时间产生影响。

6. 流量预测和资源预留：虽然自动扩容和动态重定向是有效的应对策略，但在某些情况下，预先估计流量并据此预留资源可能更为高效。

7. 平滑的拒绝策略：考虑采用回退机制，如提供降级服务、缓存结果等，这可以提供更平滑的用户体验，即使在高流量情况下。

总体上说，这是一个有前景的策略，但在实施之前需要考虑多种因素。建议在非生产环境中进行深入的测试，确保策略的有效性并优化相关的配置。

1.5 一般一个服务通常是集群部署，线程池中的队列已满，并且线程数已达到阈值时，能不能将任务存储到磁盘，等到队列不满时候再从磁盘中取出任务放入队列中？

将任务存储到磁盘是一种可能的策略，称为“磁盘交换”或“磁盘备份”。在某些场景中，这种策略可能有效，但也有一些明显的缺点和挑战：

1. I/O 性能：磁盘I/O通常比内存I/O慢得多。频繁地将任务写入磁盘和从磁盘读取任务可能导致明显的性能下降。

2. 数据持久化：一旦任务被写入到磁盘，它们会变得更为持久。如果服务器意外关机或重新启动，这些任务仍然存在，可以在重新启动后继续处理。这可能是一个优势，但也需要管理和跟踪这些存储在磁盘上的任务。

3. 磁盘空间：如果任务积压太多，可能会用尽服务器的磁盘空间，从而导致其他问题。

4. 复杂性：实施这种策略需要更复杂的编程和资源管理策略，包括任务序列化、磁盘管理和错误处理。

5. 数据一致性和完整性：磁盘故障、软件错误或其他问题可能导致存储在磁盘上的任务数据丢失或损坏。

6. 延迟：将任务移动到磁盘并在后续从磁盘中恢复，会增加处理任务的延迟。

基于上述因素，虽然将任务存储到磁盘是一种策略，但在考虑这种方法之前需要深入思考。通常更常见的策略是使用像RabbitMQ、Kafka、Redis等外部消息队列来缓冲任务，这些系统已经针对高吞吐量和数据持久化进行了优化。当内部队列满时，可以将任务放入外部消息队列，并在有空间时从外部队列中取出任务进行处理。这种策略可以有效地处理大量的任务积压，同时提供更好的扩展性和容错能力。

1.6 两个线程交替打印AB字母的实现方式

1.6.1 两个线程交替打印AB字母，有哪几种实现方式？

两个线程交替打印AB字母的几种实现方式：

1. 使用wait()和notify()方法
2. 使用Lock和Condition
3. 使用Semaphore
4. 使用AtomicInteger和自旋锁
5. 使用CountDownLatch
6. 使用CyclicBarrier
7. 使用BlockingQueue
8. 使用Exchanger

以下是对应的几种方式的Java代码示例：

1. 使用wait()和notify()方法：
   public class AlternatePrinter {

   private final Object lock = new Object();
   private int current = 1;
   private final int MAX;

   public AlternatePrinter(int max) {
   this.MAX = max;
   }

   public void printOdd() throws InterruptedException {
   synchronized (lock) {
   while (current <= MAX) {
   if (current % 2 == 0) {
   lock.wait();
   } else {
   System.out.println("Thread 1 - " + current);
   current++;
   lock.notify();
   }
   }
   }
   }

   public void printEven() throws InterruptedException {
   synchronized (lock) {
   while (current <= MAX) {
   if (current % 2 != 0) {
   lock.wait();
   } else {
   System.out.println("Thread 2 - " + current);
   current++;
   lock.notify();
   }
   }
   }
   }

   public static void main(String[] args) {
   AlternatePrinter printer = new AlternatePrinter(10);

    Thread t1 = new Thread(() -> {
        try {
            printer.printOdd();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
   
    Thread t2 = new Thread(() -> {
        try {
            printer.printEven();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
   
    t1.start();
    t2.start();
   

   }
   }

2. 使用Lock和Condition：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class PrintAB {
    private static final Lock lock = new ReentrantLock();
    private static final Condition condition = lock.newCondition();
    private static volatile boolean printA = true;

    public static void main(String[] args) {
        Thread threadA = new Thread(() -> {
            lock.lock();
            try {
                for (int i = 0; i < 10; i++) {
                    while (!printA) {
                        condition.await();
                    }
                    System.out.print("A ");
                    printA = false;
                    condition.signal();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        });

        Thread threadB = new Thread(() -> {
            lock.lock();
            try {
                for (int i = 0; i < 10; i++) {
                    while (printA) {
                        condition.await();
                    }
                    System.out.print("B ");
                    printA = true;
                    condition.signal();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        });

        threadA.start();
        threadB.start();
    }
}

3. 使用Semaphore：

import java.util.concurrent.Semaphore;

public class PrintAB {
    private static final Semaphore semaphoreA = new Semaphore(1);
    private static final Semaphore semaphoreB = new Semaphore(0);

    public static void main(String[] args) {
        Thread threadA = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    semaphoreA.acquire();
                    System.out.print("A ");
                    semaphoreB.release();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread threadB = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    semaphoreB.acquire();
                    System.out.print("B ");
                    semaphoreA.release();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadA.start();
        threadB.start();
    }
}

4. 使用AtomicInteger和自旋锁：

import java.util.concurrent.atomic.AtomicInteger;

public class PrintAB {
    private static final AtomicInteger atomicInteger = new AtomicInteger(0);

    public static void main(String[] args) {
        Thread threadA = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                while (atomicInteger.get() % 2 != 0) {
                    // Busy-wait loop (spin lock)
                }
                System.out.print("A ");
                atomicInteger.incrementAndGet();
            }
        });

        Thread threadB = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                while (atomicInteger.get() % 2 == 0) {
                    // Busy-wait loop (spin lock)
                }
                System.out.print("B ");
                atomicInteger.incrementAndGet();
            }
        });

        threadA.start();
        threadB.start();
    }
}

5. 使用CountDownLatch交替打印的代码示例：

import java.util.concurrent.CountDownLatch;

public class PrintAB {
    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            CountDownLatch latchA = new CountDownLatch(1);
            CountDownLatch latchB = new CountDownLatch(1);

            Thread threadA = new Thread(() -> {
                try {
                    System.out.print("A ");
                    latchB.countDown();
                    latchA.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            Thread threadB = new Thread(() -> {
                try {
                    latchB.await();
                    System.out.print("B ");
                    latchA.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            threadA.start();
            threadB.start();
        }
    }
}

6. 使用CyclicBarrier交替打印的代码示例：

import java.util.concurrent.CyclicBarrier;

public class PrintAB {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(2);

        Thread threadA = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.print("A ");
                    barrier.await();
                    barrier.await();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        Thread threadB = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    barrier.await();
                    System.out.print("B ");
                    barrier.await();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        threadA.start();
        threadB.start();
    }
}

7. 使用BlockingQueue交替打印的代码示例：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class PrintAB {
    public static void main(String[] args) {
        BlockingQueue<Character> queueA = new ArrayBlockingQueue<>(1);
        BlockingQueue<Character> queueB = new ArrayBlockingQueue<>(1);

        Thread threadA = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.print("A ");
                    queueB.put('B');
                    queueA.take();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread threadB = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    queueB.take();
                    System.out.print("B ");
                    queueA.put('A');
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadA.start();
        threadB.start();
    }
}

1.6.2 追问：核心思路要解决的问题？

加锁然后判断（原子类）
2. 核心思路：
首先需要实现两个线程的同步，即当一个线程打印A后，要通知另一个线程开始打印B，打印完B后再通知第一个线程打印A。这样就实现了两个线程的交替打印。
需要解决的问题是如何实现两个线程的同步和通信，如何确保当一个线程执行完毕后另一个线程能够立即开始执行。

1.6.3 追问：可以用wait、notify、join、yield吗？

可以使用wait、notify、join、yield吗？

• wait和notify可以用来实现线程的同步和通信，可以用于此场景。
• join可以用来使一个线程等待另一个线程完成，但不适合此场景，因为我们需要两个线程交替执行。
• yield可以使当前线程让出CPU执行权，但不能保证立即将执行权交给指定的线程，所以不适合此场景。

以下是使用wait和notify实现的示例代码：

public class AlternatePrint {

    private static final Object lock = new Object();

    public static void main(String[] args) {
        Thread threadA = new Thread(() -> {
            synchronized (lock) {
                for (int i = 0; i < 10; i++) {
                    System.out.print("A");
                    lock.notify();
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                lock.notify(); // Ensure the other thread can exit the loop
            }
        });

        Thread threadB = new Thread(() -> {
            synchronized (lock) {
                for (int i = 0; i < 10; i++) {
                    System.out.print("B");
                    lock.notify();
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                lock.notify(); // Ensure the other thread can exit the loop
            }
        });

        threadA.start();
        threadB.start();
    }
}

此代码中，两个线程分别打印A和B，并使用wait和notify进行同步和通信，确保两个线程可以交替打印。文章来源地址https://www.toymoban.com/news/detail-628831.html

到了这里，关于深度思考线程池面经的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！