Java의 Thread Pool과 Executor의 원리에 대한 자세한 설명

이 글은 주로 Java 스레드 풀 분석과 Executor 원리를 자세히 설명하는 관련 정보를 소개합니다. 이 부분의 지식을 모두가 이해할 수 있도록 예제와 분석 원리를 제공합니다.

자세한 설명은 다음과 같습니다. Java 스레드 풀 및 실행자 원리 분석

스레드 풀 기능 및 기본 지식

시작하기 전에 먼저 "스레드 풀"의 개념에 대해 논의하겠습니다. "스레드 풀"은 이름에서 알 수 있듯이 스레드 캐시입니다. 하나 이상의 스레드 집합으로, 사용자는 실행 세부 사항에 너무 얽매이지 않고 수행해야 할 작업을 스레드 풀에 간단히 던질 수 있습니다. 그렇다면 스레드 풀의 기능은 무엇입니까? 아니면 Thread를 직접 사용하는 것에 비해 어떤 이점이 있나요? 다음 사항을 간략하게 정리했습니다.

스레드 생성 및 소멸로 인한 소비 감소

Java Thread 구현에 대해서는 이전 블로그에서 분석한 적이 있습니다. Java Thread와 커널 스레드는 1:1입니다(Linux). 또한 Thread는 Java 계층과 C++ 계층에 멤버 데이터가 많기 때문에 실제로 Java Thread는 상대적으로 무겁습니다. Java Thread를 생성하고 삭제하려면 OS와 JVM 모두 많은 작업을 수행해야 하므로 Java Thread를 캐시하면 어느 정도 효율성이 향상될 수 있습니다.

컴퓨팅 리소스 제어의 더욱 편리하고 투명한 구현

이 항목을 논의하려면 몇 가지 예를 들어야 할 수도 있습니다. 매우 유명한 웹 서버 Nginx를 예로 들어보겠습니다. Nginx는 강력한 동시성 기능과 낮은 리소스 소비로 유명합니다. 이러한 엄격한 요구 사항을 충족하기 위해 Nginx는 작업자 스레드 수를 엄격하게 제한합니다(작업자 스레드는 일반적으로 CPU 수와 동일합니다). 이 설계의 초점은 스레드 전환으로 인한 성능 손실을 줄이는 것입니다. 이 최적화 방법은 Java에도 적용 가능합니다. 각 작업마다 새로운 스레드가 생성되면 최종 결과는 프로그램 리소스를 제어하기 어렵고(특정 기능이 CPU를 가득 채우는 현상) 전반적인 실행 속도가 상대적으로 느려지는 것입니다. Java 스레드 풀은 최대 스레드 수를 제어하는 ​​데 사용할 수 있는 FixThreadPool을 제공합니다.

위에 언급된 "말도 안 되는" 내용이 너무 많아서 Java 스레드 풀 구현을 기반으로 분석해 보겠습니다! Java의 스레드 풀에는 여러 가지 구현이 있습니다.

cached ThreadPool

캐시된 스레드 풀의 특징은 이전 스레드를 캐시하고 새로 제출된 작업이 캐시된 스레드에서 실행될 수 있다는 것입니다. 이는 위에서 언급한 내용을 달성합니다. 첫 번째 장점 .

fixed ThreadPool

cachedThreadPool의 한 가지 기능은 새로 제출된 작업을 실행할 유휴 스레드가 없으면 새 스레드가 생성된다는 것입니다. FixThreadPool은 이 작업을 수행하지 않고 작업을 저장하고 실행하기 전에 유휴 스레드가 있을 때까지 기다립니다. 즉, 위에서 언급한 두 번째 이점이 달성됩니다.

scheduled ThreadPool

예약 ThreadPool의 특징은 작업의 지연 실행, 주기적인 작업 실행 등 작업 스케줄링을 실현할 수 있다는 것입니다.

위의 세 가지 외에도 Java는 Fork/Join 프레임워크를 기반으로 하는 newWorkStealingPool도 구현합니다. 아직 이것에 대해 살펴보지 않았으므로 그대로 두겠습니다. Java의 동시성 지원에서 Executor는 다양한 스레드 풀을 패키지하는 데 사용됩니다. "executor"라는 이름은 실제로는 실행자일 뿐입니다.

1.캐시된 ThreadPool과 고정 ThreadPool의 구현

이전 설명에서 볼 수 있듯이 이 두 스레드 풀은 매우 유사합니다. 실제로는 동시에 구현됩니다. 그렇지 않은 경우 실제 예를 살펴보겠습니다.

ThreadPoolExecutor executor1 = (ThreadPoolExecutor)Executors.newCachedThreadPool();

ThreadPoolExecutor executor2 = (ThreadPoolExecutor)Executors.newFixedThreadPool(4);

이것은 두 개의 스레드 풀을 만드는 새로운 방법입니다. 매우 유사합니다! 그렇게 생각하지 않는다면 나는 진실만을 보여줄 수 있을 뿐입니다.

public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                 60L, TimeUnit.SECONDS,
                 new SynchronousQueue<Runnable>());
}

public static ExecutorService newFixedThreadPool(int nThreads) {
  return new ThreadPoolExecutor(nThreads, nThreads,
                 0L, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<Runnable>());
}

예, 매개변수만 약간 다를 뿐 동일한 생성자를 호출합니다. 이제 이러한 매개변수의 의미와 두 매개변수 세트 간의 차이점을 살펴보겠습니다. 먼저 ThreadPoolExecutor의 생성자를 게시해야 합니다.

public ThreadPoolExecutor(int corePoolSize,
             int maximumPoolSize,
             long keepAliveTime,
             TimeUnit unit,
             BlockingQueue<Runnable> workQueue) {
  this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
     Executors.defaultThreadFactory(), defaultHandler);
}

새롭게 보이기 위해 다른 레이어의 생성자를 게시하지 않을 것이며 해당 생성자는 단순한 할당일 뿐입니다. 여기의 함수 프로토타입은 이미 많은 정보를 제공할 수 있습니다. JDK 코드 이름 지정은 주석과 마찬가지로 정말 좋습니다.

maximumPoolSize는 스레드 풀의 최대 스레드 수입니다. 캐시된 ThreadPool의 경우 이 값은 기본적으로 수십억 개의 스레드를 실행할 수 있는 Integer.MAX_VALUE입니다. ! 고정 ThreadPool의 경우 이 값은 사용자가 설정한 스레드 풀 수입니다.
keepAliveTime 및 단위는 스레드의 캐시 만료 시간을 결정합니다. 캐시된 ThreadPool의 경우 스레드의 캐시 만료 시간은 1분입니다. 즉, 작업자 스레드가 1분 동안 아무 작업도 하지 않으면 저장을 위해 취소됩니다. 자원. 고정 ThreadPool에 전달된 시간은 0입니다. 이는 고정 ThreadPool의 작업자 스레드가 만료되지 않음을 의미합니다.

corePoolSize是线程池的最小线程数；对于cached ThreadPool，这个值为0，因为在完全没有任务的情况下，cached ThreadPool的确会成为“光杆司令”。至于fixed ThreadPool，这个fixed已经表明corePoolSize是等于线程总数的。
接下来，我们根据一个简单的使用例子，来看看一下cached ThreadPool的流程。

public class Task implements Callable<String> {

private String name;
public Task(String name) {
  this.name = name;
}
@Override
public String call() throws Exception {
  System.out.printf("%s: Starting at : %s\n", this.name, new Date());
  return "hello, world";
}
public static void main(String[] args) {
  ThreadPoolExecutor executor = (ThreadPoolExecutor)Executors.newCachedThreadPool();
  Task task = new Task("test");
  Future<String> result = executor.submit(task);
  try {
    System.out.printf("%s\n", result.get());
  } catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
  }
  executor.shutdown();
  System.out.printf("Main ends at : %s\n", new Date());
}
}

首先，来看看executor.submit(task)，这其实调用了ThreadPoolExecutor.execute(Runnable command)方法，这个方法的代码如下，整段代码的逻辑是这样的。首先检查线程池的线程数是否不够corePoolSize，如果不够就直接新建线程并把command添加进去；如果线程数已经够了或者添加失败（多个线程增加添加的情况），就尝试把command添加到队列中（workQueue.offer(command)），如果添加失败了，就reject掉cmd。大体的逻辑是这样的，这段代码有很多基于线程安全的设计，这里为了不跑题，就先忽略细节了。

public void execute(Runnable command) {
  if (command == null)
    throw new NullPointerException();
  int c = ctl.get();
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
      return;
    c = ctl.get();
  }
  if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    if (! isRunning(recheck) && remove(command))
      reject(command);
    else if (workerCountOf(recheck) == 0)
      addWorker(null, false);
  }
  else if (!addWorker(command, false))
    reject(command);
}

到这里，看起来线程池实现的整体思路其实也没多么复杂。但是还有一个问题——一个普通的Thread在执行完自己的run方法后会自动退出。那么线程池是如何实现Worker线程不断的干活，甚至在没有任务的时候。其实答案很简单，就是Worker其实在跑大循环，Worker实际运行方法如下：

final void runWorker(Worker w) {
  Thread wt = Thread.currentThread();
  Runnable task = w.firstTask;
  w.firstTask = null;
  w.unlock(); // allow interrupts
  boolean completedAbruptly = true;
  try {
    while (task != null || (task = getTask()) != null) {
      w.lock();
  /***/
      try {
        beforeExecute(wt, task);
        Throwable thrown = null;
        try {
          task.run();
        /***/
        } finally {
          afterExecute(task, thrown);
        }
      } finally {
        task = null;
        w.completedTasks++;
        w.unlock();
      }
    }
    completedAbruptly = false;
  } finally {
    processWorkerExit(w, completedAbruptly);
  }
}

关键就在这个while的判断条件，对于需要cached线程的情况下，getTask()会阻塞起来，如果缓存的时间过期，就会返回一个null，然后Worker就退出了，也就结束了它的服役周期。而在有任务的情况下，Woker会把task拿出来，然后调用task.run()执行任务，并通过Future通知客户线程（即future.get()返回）。这样一个简单的线程池使用过程就完了。。。

当然，线程池的很多精髓知识——基于线程安全的设计，我都没有分析。有兴趣可以自己分析一下，也可以和我讨论。此外Scheduled ThreadPool这里也没有分析，它的要点其实是调度，主要是根据时间最小堆来驱动的。

위 내용은 Java의 Thread Pool과 Executor의 원리에 대한 자세한 설명의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!