Java 스레드 풀 실행자 사용 방법

스레드 풀 클래스 다이어그램

스레드 풀을 생성하고 스레드를 사용하기 위해 가장 일반적으로 사용되는 Executor 구현은 주로 위의 클래스 다이어그램에서 제공되는 클래스를 사용합니다. 위의 클래스 다이어그램에는 실행을 예약하고 일련의 실행 전략 호출을 기반으로 비동기 작업을 제어하는 ​​프레임워크인 Executor 프레임워크가 포함되어 있습니다. 목적은 작업 실행 방법과 작업 제출을 분리하는 메커니즘을 제공하는 것입니다. 여기에는 세 가지 실행기 인터페이스가 포함되어 있습니다.

• Executor: 새 작업을 실행하기 위한 간단한 인터페이스

• ExecutorService: Executor를 확장하고 실행기 수명 주기 및 작업 수명 주기를 관리하는 메서드 추가

• ScheduleExcutorService: 지원하는 확장 ExecutorService 향후 및 주기적 작업 실행

스레드 풀의 이점

• 리소스 소비 감소 - 기존 스레드 재사용, 객체 생성 및 종료 비용 감소, 성능 향상

• 응답 속도 - It 최대 동시 스레드 수를 효과적으로 제어하고, 시스템 리소스 활용도를 향상시키며, 동시에 과도한 리소스 경쟁을 방지하고 차단을 방지할 수 있습니다. 태스크가 도착하면 스레드 생성을 기다리지 않고 즉시 태스크를 실행할 수 있습니다

• 스레드 관리성 향상 - 예약 실행, 주기 실행, 단일 스레드, 동시성 제어 등의 기능을 제공합니다.

새 스레드의 단점

• 새 스레드가 새 객체를 생성할 때마다 성능이 저하됩니다.

• 통합 스레드 관리가 부족하여 무제한의 새 스레드가 생겨 서로 경쟁하고 점유할 수 있습니다. 시스템 리소스가 너무 많아 충돌이나 OOM(메모리 부족 메모리 오버플로)이 발생하는 경우, 이 문제의 원인은 단순히 새로운 스레드가 아니라 프로그램 버그나 설계 결함으로 인해 지속적으로 새로운 스레드가 발생할 수도 있습니다.

• 추가 실행, 주기적 실행, 스레드 중단과 같은 추가 기능이 없습니다.

스레드 풀 코어 클래스-ThreadPoolExecutor

매개변수 설명: ThreadPoolExecutor에는 총 7개의 매개변수가 함께 작동하여 스레드 풀의 강력한 기능을 형성합니다.

corePoolSize: 코어 스레드 수

maximumPoolSize: 최대 스레드 수

workQueue: 대기열 차단, 실행 대기 중인 작업 저장, 매우 중요합니다. 스레드에 상당한 영향을 미칩니다. 풀 실행 프로세스

새 작업을 스레드 풀에 제출하면 스레드 풀은 현재 풀에서 실행 중인 스레드 수에 따라 작업을 처리하는 방법을 결정합니다. 3가지 처리 방법이 있습니다.

1. 직접 전환(SynchronusQueue)

2. Unbounded queue(LinkedBlockingQueue) 생성할 수 있는 최대 스레드 수는 corePoolSize이며, 이때 maximumPoolSize는 작동하지 않습니다. 스레드 풀의 모든 코어 스레드가 실행 중이면 새 작업 제출이 대기 대기열에 배치됩니다.

3. 제한된 대기열(ArrayBlockingQueue)의 최대 풀 크기는 리소스 소비를 줄일 수 있지만 이 방법은 스레드 풀 예약을 더 어렵게 만듭니다. 스레드 풀과 대기열 용량이 제한되어 있기 때문입니다. 따라서 스레드 풀 및 처리 작업의 처리 속도가 합리적인 범위에 도달하기를 원하고 스레드 스케줄링을 상대적으로 단순하게 만들고 리소스 소비를 최대한 줄이려면 이 두 가지 수량 할당 기술을 합리적으로 제한해야 합니다. : [ CPU 사용량, 운영 체제 리소스 사용량, 컨텍스트 전환 오버헤드 등을 포함하여 리소스 사용량을 줄이려는 경우 대기열 용량을 크게 설정하고 스레드 풀 용량을 작게 설정하면 스레드 풀의 처리량이 줄어듭니다. . 제출하는 작업이 자주 차단되는 경우 maximumPoolSize를 조정할 수 있습니다. 대기열 용량이 작다면 스레드 풀 크기를 더 크게 설정해야 CPU 사용량이 상대적으로 높아집니다. 그러나 스레드 풀의 용량을 너무 크게 설정하여 작업 개수를 너무 많이 늘리면 동시성(concurrency)의 양이 늘어나게 되므로 스레드 간 스케줄링을 고려해야 할 문제이다. 대신 처리 작업의 처리량이 줄어들 수 있습니다. ]

keepAliveTime: 작업 실행이 없을 때 스레드가 지속될 수 있는 시간(스레드의 스레드 수가 corePoolSize보다 큰 경우, 이때 제출된 새 작업이 없으면 코어 스레드 외부의 스레드) 즉시 소멸되지 않고 keepAliveTime이 초과될 때까지 기다립니다)

unit: keepAliveTime의 시간 단위

threadFactory: 스레드를 생성하는 데 사용되는 스레드 팩토리, 스레드를 생성하는 기본 팩토리가 있으므로 새로 생성된 스레드의 우선순위가 같다는 것, 데몬이 아닌 스레드인지, 이름이 설정되어 있는지)

rejectHandler: 작업 처리를 거부할 때의 정책(차단 대기열이 가득 찼을 때)(AbortPolicy 기본 정책은 예외가 직접 발생하면 CallerRunsPolicy는 호출자의 스레드를 사용하여 작업을 실행하고 DiscardOldestPolicy는 대기열을 삭제합니다. 상위 작업과 현재 작업을 실행하면 DiscardPolicy는 현재 작업을 직접 삭제합니다.

corePoolSize, maximumPoolSize 및 workQueue의 관계: 실행 중인 스레드 수가 corePoolSize보다 적으면 작업을 처리하기 위해 새 스레드가 직접 생성됩니다. 스레드 풀의 다른 스레드가 유휴 상태인 경우에도 마찬가지입니다. 실행 중인 스레드 수가 corePoolSize보다 크고 maximumPoolSize보다 작은 경우 workQueue가 가득 찼을 때만 작업을 처리하기 위해 새 스레드가 생성됩니다. corePoolSize와 maximumPoolSize가 동일한 경우 생성되는 스레드 풀의 크기는 고정됩니다. 이때 새로운 작업이 제출되며, workQueue가 가득 차지 않으면 요청이 workQueue에 배치됩니다. 빈 스레드가 작업 대기열에서 작업을 제거할 때까지 기다립니다. 이때 workQueue도 가득 찬 경우 추가 거부 정책 매개변수를 사용하여 거부 정책을 실행합니다.

초기화 방법: 7개의 매개변수를 4개의 초기화 방법으로 구성

기타 방법:

execute();	//提交任务，交给线程池执行	
submit();//提交任务，能够返回执行结果 execute+Future
shutdown();//关闭线程池，等待任务都执行完
shutdownNow();//关闭线程池，不等待任务执行完
getTaskCount();//线程池已执行和未执行的任务总数
getCompleteTaskCount();//已完成的任务数量
getPoolSize();//线程池当前的线程数量
getActiveCount();//当前线程池中正在执行任务的线程数量

스레드 풀 수명 주기:

• running: 새로 제출된 작업을 수락하고 처리할 수도 있습니다. the Blocking queue

• shutdown: 새 작업을 처리할 수 없지만 차단 대기열에 있는 작업은 계속 처리할 수 있습니다

• stop: 새 작업을 받을 수도 없고 대기열에 있는 작업도 처리할 수 없습니다

• tidying: 모든 작업이 종료되었으며 유효 스레드 수는 0

• 종료됨: 최종 상태

Executor를 사용하여 스레드 풀 생성

Executor를 사용하면 위에서 언급한 스레드 풀에 해당하는 4개의 스레드 풀을 생성할 수 있습니다. 풀 초기화 방법

Executors.newCachedThreadPool

newCachedThreadPool은 작업이 제출되면 corePoolSize가 0이고 코어 스레드가 생성되지 않는 스레드 풀입니다. 이해할 수 있음 대기열이 항상 가득 차 있기 때문에 결국 작업을 수행하기 위해 비핵심 스레드가 생성됩니다. 비코어 스레드는 60초 동안 유휴 상태일 때 재활용됩니다. Integer.MAX_VALUE는 크기가 매우 크기 때문에 스레드를 무한히 생성할 수 있어 리소스가 제한되어 있을 경우 OOM 예외가 쉽게 발생할 수 있다고 생각할 수 있습니다.

//创建newCachedThreadPool线程池源码
public static ExecutorService newCachedThreadPool() {
		/**
        *corePoolSize： 0，核心线程池的数量为0
		*maximumPoolSize：  Integer.MAX_VALUE，可以认为最大线程数是无限的
		*keepAliveTime： 60L
		*unit: 秒
		*workQueue: SynchronousQueue
        **/
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

사용 사례:

public static void main(String[] args) {
    ExecutorService executor = Executors.newCachedThreadPool();
    for (int i = 0; i < 10; i++) {
        final int index = i;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                log.info("task:{}",index);
            }
        });
    }
}

newCachedThreadPool의 반환 값은 ExecutorService 유형이라는 점에 유의할 필요가 있습니다. 이 유형에는 기본 스레드 풀 메서드만 포함되어 있으며, 스레드 모니터링 관련 메서드는 포함되지 않습니다. 새 스레드를 생성할 때 ExecutorService 스레드 풀 유형을 고려해야 합니다.

Executors.newSingleThreadExecutor

newSingleThreadExecutor는 단일 스레드 풀로, 단 하나의 코어 스레드만 갖고 작업을 실행하는 데 단 하나의 공유 스레드를 사용하여 모든 작업이 지정된 순서(FIFO, 우선 순위)로 실행되도록 합니다. ..)

//newSingleThreadExecutor创建线程池源码
public static ExecutorService newSingleThreadExecutor() {
    /**
      *  corePoolSize : 1，核心线程池的数量为1

      *  maximumPoolSize : 1，只可以创建一个非核心线程

      *  keepAliveTime : 0L

      *  unit => 秒

      *  workQueue => LinkedBlockingQueue
      **/
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

작업이 제출되면 작업을 실행하기 위해 먼저 코어 스레드가 생성됩니다. 코어 스레드 수가 이 수를 초과하면 LinkedBlockingQueue는 길이가 있는 대기열이므로 대기열에 추가됩니다. Integer.MAX_VALUE의 경우 무한한 대기열로 간주될 수 있으므로 대기열에 무제한으로 삽입할 수 있으며, 이는 무제한으로 인해 리소스가 제한될 때 쉽게 OOM 예외가 발생할 수 있습니다. 대기열의 경우 maximumPoolSize 및 keepAliveTime 매개변수는 유효하지 않으며 비핵심 스레드는 전혀 생성되지 않습니다.

Executors.newFixedThreadPool

고정 길이 스레드 풀, 코어 스레드 수 및 최대 스레드 수가 사용자에 의해 전달됩니다. 최대 동시 스레드 수를 설정할 수 있으며 최대 동시 스레드 수는 대기합니다. in the queue

//newFixedThreadPool创建线程池源码
public static ExecutorService newFixedThreadPool(int nThreads) {
    	/**
          *  corePoolSize : 核心线程的数量为自定义输入nThreads

          *  maximumPoolSize ： 最大线程的数量为自定义输入nThreads

          *  keepAliveTime ： 0L

          *  unit ： 秒

          *  workQueue ： LinkedBlockingQueue
          **/
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

newFixedThreadPool은 SingleThreadExecutor와 유사하지만 유일한 차이점은 코어 스레드의 숫자가 다르며 LinkedBlockingQueue를 사용하기 때문에 리소스가 제한될 때 OOM 예외가 발생하기 쉽습니다.

Executors.newScheduledThreadPool

고정 길이 스레드 풀, 사용자가 코어 스레드 수 전달, 예약 및 주기적인 작업 실행 지원

//newScheduledThreadPool创建线程池源码
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    /**
      *  corePoolSize : 核心线程的数量为自定义输入corePoolSize

      *  maximumPoolSize ： 最大线程的数量为Integer.MAX_VALUE

      *  keepAliveTime ： 0L

      *  unit ： 纳秒

      *  workQueue ： DelayedWorkQueue
      **/
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

작업이 제출되면 corePoolSize가 사용자 정의 입력이고 코어 스레드가 생성됩니다. 먼저 코어 스레드가 가득 차서 결국에는 작업을 수행하기 위해 비코어 스레드가 생성됩니다. 비핵심 스레드는 사용 후 재활용됩니다. Integer.MAX_VALUE는 크기가 매우 크기 때문에 스레드를 무한히 생성할 수 있어 리소스가 제한되어 있을 경우 OOM 예외가 쉽게 발생할 수 있다고 생각할 수 있습니다. 사용된 DelayedWorkQueue는 예약된 작업과 주기적인 작업을 구현할 수 있기 때문입니다. ScheduledExecutorService는 사용할 수 있는 세 가지 방법을 제공합니다.

schedule: 지연 후 작업 실행 ScheduleAtFixedRate: 지정된 속도로 작업 실행 ScheduleWithFixedDelay: 지정된 지연으로 작업 실행 사용 사례:

    public static void main(String[] args) {

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

//        executorService.schedule(new Runnable() {
//            @Override
//            public void run() {
//                log.warn("schedule run");
//            }
//         //延迟3秒后执行
//        }, 3, TimeUnit.SECONDS);
        //        executorService.shutdown();

//        executorService.scheduleWithFixedDelay(new Runnable() {
//            @Override
//            public void run() {
//                log.warn("scheduleWithFixedDelay run");
//            }
//            //延迟一秒后每隔3秒执行
//        }, 1, 3, TimeUnit.SECONDS);
        
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                log.warn("schedule run");
            }
            //延迟一秒后每隔3秒执行
        }, 1, 3, TimeUnit.SECONDS);

        /**
         * 定时器调度，不推荐使用，推荐ScheduledExecutorService调度
         */
//        Timer timer = new Timer();
//        timer.schedule(new TimerTask() {
//            @Override
//            public void run() {
//                log.warn("timer run");
//            }
//        //从当前时间每隔5秒执行
//        }, new Date(), 5 * 1000);
    }

요약

• FixThreadPool 및 SingleThreadExecutor 허용되는 요청 대기열 길이는 Integer.MAX_VALUE이며, 이로 인해 많은 수의 요청이 누적되어 OOM 예외가 발생할 수 있습니다

• CachedThreadPool 및 newScheduledThreadPool에서 생성할 수 있는 스레드 수는 Integer.MAX_VALUE이며, 이로 인해 많은 수의 스레드가 생성되어 OOM 예외가 발생할 수 있습니다.

이것이 Executor를 사용하여 스레드 풀을 생성하는 것을 금지하는 이유입니다. 하지만 ThreadPoolExecutor를 직접 생성하는 것이 좋습니다. 이유

스레드 풀 매개 변수 정의 방법

CPU 집약적: 스레드 풀의 크기는 CPU 수 + 1을 권장합니다. CPU 수를 얻을 수 있습니다. Runtime.availableProcessors 메소드에 따라 IO 집약적: CPU 수 * CPU 사용률 * ( 1 + 스레드 대기 시간/스레드 CPU 시간) 하이브리드 유형: 작업을 CPU 집약적 및 IO 집약적으로 나누어 사용 Blocking queue 조정: 제한된 대기열을 사용하는 것이 리소스 고갈을 방지하는 데 도움이 됩니다. 거부 전략: 기본값은 다음과 같습니다. 프로그램에서 직접 RejectedExecutionException을 발생시키는 AbortPolicy 거부 전략(시간 예외가 실행 중이기 때문에 강제 포착 없음), 이 처리 방법은 충분히 우아하지 않습니다. 거부 처리에는 다음 전략이 권장됩니다.

• 프로그램에서 RejectedExecutionException 예외를 포착하고 포착된 예외에서 작업을 처리합니다. 기본 거부 정책

• 의 경우 CallerRunsPolicy 거부 정책을 사용합니다. 이 정책은 실행을 호출하는 스레드(보통 메인 스레드)에 작업을 넘겨줍니다. 이때 기본 스레드는 어떤 작업도 제출할 수 없습니다. 일정 기간 동안 작업자 스레드가 진행 중인 작업을 처리하게 합니다. 이때 제출된 스레드는 TCP 대기열에 저장됩니다. TCP 대기열이 가득 차면 이는 클라이언트에 영향을 미치게 됩니다. 이는 거부 전략을 사용자 정의하려면 RejectedExecutionHandler 인터페이스만 구현하면 됩니다.

• 작업이 특별히 중요하지 않은 경우 DiscardPolicy 및 DiscardOldestPolicy 거부 정책을 사용하여 작업을 삭제하는 것도 가능합니다.

• Executors의 정적 메서드를 사용하여 ThreadPoolExecutor 개체를 생성하는 경우 Semaphore를 사용할 수 있습니다. 작업 실행을 제한하고 OOM 예외를 방지하려면

위 내용은 Java 스레드 풀 실행자 사용 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!