[Java 동시 실행] - J.U.C의 스레드 풀:ScheduledThreadPoolExecutor

[Java 동시 실행] - J.U.C의 스레드 풀:ScheduledThreadPoolExecutor
 
 
ScheduledThreadPoolExecutor 해석
우리는 Timer와 TimerTask가 비록 라인의 주기와 지연 스케줄링을 실현할 수 있지만 Timer와 TimerTask는 약간의 결함이 존재한다는 것을 알고 있기 때문에 이러한 정기적이고 주기적으로 임무를 수행하는 스케줄링 전략에 대해 우리는 보통ScheduledThreadPoolExecutor를 추천하여 실현한다.다음은ScheduledThreadPoolExecutor가 어떻게 라인의 주기, 지연 스케줄링을 실현하는지 깊이 있게 분석한다.Scheduled Thread Pool Executor는 Thread Pool Executor를 계승하고 Scheduled Executor 서비스 인터페이스를 실현했다. 이것은'지연'과'주기 집행'기능을 제공하는 Thread Pool Executor에 해당한다.JDK API에서 이렇게 정의합니다: ThreadPoolExecutor. 지정한 지연 후에 명령을 실행하거나 정기적으로 명령을 실행할 수 있습니다.여러 개의 보조 스레드가 필요하거나 ThreadPoolExecutor가 추가 유연성이나 기능을 요구할 때 Timer보다 우수합니다.지연된 작업이 활성화되면 실행되지만, 언제 활성화되고, 활성화된 후에 실행되는지는 실시간으로 보장되지 않습니다.제출한 선진 선출 (FIFO) 순서에 따라 같은 실행 시간에 배정된 작업을 사용합니다.다음과 같은 네 가지 구성 방법을 제공합니다.

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue());
    }

    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue(), threadFactory);
    }

    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue(), handler);
    }


    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue(), threadFactory, handler);
    }

물론 우리는 그 구조 함수를 통해ScheduledThreadPoolExecutor 대상(예를 들어 new ScheduledThreadPoolExecutor(10)과 같은)을 직접 생성하지 않고Executors 클래스(예를 들어Executors. new ScheduledThreadPool(int)를 통해)Scheduled Thread Pool Executor의 구조 함수에서 우리는 이것이Thread Local Executor를 이용하여 구성된 것을 발견했다. 유일한 변동은 그것이 사용하는 막힌 대기열이 Delayed Work Queueue로 바뀌었기 때문이지Thread Localh Executor의 Linked Blocking Queue(Executors를 통해 Thread Localh Executor 대상을 생성하는 것)가 아니라는 것이다.DelayedWorkQueue는ScheduledThreadPoolExecutor의 내부 클래스입니다. 대기열을 막는 DelayQueue와 약간 유사합니다.DelayQueue는 지연된 차단 대기열을 제공합니다. 지연이 만료되었을 때만 원소를 추출할 수 있으며, 그 헤더는 지연이 만료된 후에 저장된 시간이 가장 긴 Delayed 원소입니다.만약 지연이 아직 만료되지 않았다면, 대기열에 머리가 없고,poll은null로 되돌아갈 것입니다.DelayQueue에 대한 더 많은 소개는 이 블로그 [Java 병발]---J.U.C의 막힌 대기열:DelayQueue를 참고하세요.그래서 DelayedWorkQueue의 임무는 지연 시간에 따라 짧고 길게 정렬됩니다.다음은 DelayedWorkQueue를 깊이 있게 분석한 다음에 여기에 인용문을 남깁니다.ScheduledThreadPoolExecutor는 다음과 같은 네 가지 방법을 제공합니다. 즉, 네 개의 스케줄러입니다.

schedule(Callable callable, long delay, Time Unit unit): 지정된 지연 후에 활성화된ScheduledFuture를 만들고 실행합니다.

schedule(Runnable command, long delay, Time Unit unit): 지정된 지연 후에 사용할 일회성 작업을 만들고 실행합니다.

scheduleAtFixedRate(Runnable command, long initial Delay, long period, Time Unit unit): 주어진 초기 지연 후 처음으로 사용되는 정기 조작을 만들고 실행하며 후속 조작은 주어진 주기를 가진다.즉, initial Delay 이후에 실행을 시작하고 initial Delay +period 이후에 실행한 다음에 initial Delay + 2 *period 이후에 실행을 하는 것으로 추정된다.

scheduleWithFixedDelay(Runnable command, long initial Delay, long delay, Time Unit unit): 지정된 초기 지연 후에 처음으로 사용되는 정기 동작을 만들고 실행합니다. 그 다음에 실행 종료와 다음 실행 시작 사이에 지정된 지연이 존재합니다.

첫 번째, 두 번째 방법은 차이가 많지 않고 모두 일회성 조작이다. 단지 매개 변수 하나는 Callable이고 하나는 Runnable이다.세 번째(schedule AtFixed Rate), 네 번째(schedule With Fixed Delay) 방법을 살짝 분석하고 initial Delay = 5,period/delay = 3,unit를 초로 넣는다.만약 모든 스레드가 매우 잘 운행되고 지연되는 문제가 존재하지 않는다면 이 두 방법의 스레드 운행 주기는 5, 8, 11, 14, 17......그런데 딜레이가 있다면?예를 들어 세 번째 라인이 5초 걸렸다면 이 두 가지 방법의 처리 전략은 어떤 것입니까?세 번째 방법(schedule AtFixedRate)은 주기가 고정되어 있다. 즉, 이 지연의 영향을 받지 않는다는 것이다. 모든 라인의 스케줄링 주기는 초기화할 때 이미 절대적이다. 언제 스케줄링을 하면 언제 스케줄링을 하고 이전 라인의 스케줄링 실효 지연으로 인해 영향을 받지 않는다.그러나 네 번째 방법(schedule With Fixed Delay)은 다르다. 이것은 각 라인의 스케줄링 간격이 고정된 것이다. 즉, 첫 번째 라인과 두 번째 라인 사이의 간격은delay, 두 번째 라인과 세 번째 간격은delay로 추정된다.만약에 두 번째 라인이 늦어진다면 뒤에 있는 모든 라인의 스케줄이 늦어진다. 예를 들어 위의 두 번째 라인이 2초 늦어진다면 세 번째 라인은 11초가 아니라 13초가 된다.네 가지 방법의 원본을 살펴보면 사실 그들의 처리 논리에 차이가 많지 않다는 것을 알 수 있기 때문에 우리는 scheduleWithFixedDelay 방법을 골라 분석한다. 다음과 같다.

    public ScheduledFuture> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (delay <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask sft =
            new ScheduledFutureTask(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));
        RunnableScheduledFuture t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

scheduleWithFixedDelay 메소드의 논리는 다음과 같습니다.

검사, 매개 변수가 법칙에 맞지 않으면 이상을 던집니다

ScheduledFutureTask

인task 구성

delayedExecute () 방법을 호출하여 후속 관련 처리를 한다

이 코드는 두 가지 종류인 Scheduled Future Task와 Runnable Scheduled Future와 Runnable Scheduled Future와 Scheduled Future 두 인터페이스를 계승한다. 그 중에서 Runnable Future와 Scheduled Future 두 가지 특성을 가진 것 외에 이 방법은 isPeriodic()를 정의했다. 이 방법은 실행하는 작업이 정기적인 작업인지 아닌지를 판단하는 데 사용되고 만약true로 되돌아온다.한편, Scheduled Future Task는Scheduled Thread Pool Executor의 내부 클래스로서 매우 중요한 역할을 한다. 왜냐하면 그의 역할은Scheduled Thread Pool Executor의 임무 스케줄링을 담당하기 때문이다.ScheduledFutureTask 내부는 FutureTask를 계승하여 RunnableScheduledFuture 인터페이스를 실현하고 내부에 세 가지 비교적 중요한 변수를 정의한다.

        /**       ScheduledThreadPoolExecutor     */
        private final long sequenceNumber;

        /**            */
        private long time;

        /**          */
        private final long period;

이 세 변수는 임무의 집행과 매우 밀접한 관계를 가지는데, 어떤 관계입니까?먼저 ScheduledFutureTask의 몇 가지 구조 함수와 핵심 방법을 살펴보겠습니다.

        ScheduledFutureTask(Runnable r, V result, long ns) {
            super(r, result);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        ScheduledFutureTask(Callable callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        ScheduledFutureTask(Callable callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

ScheduledFutureTask는 위의 세 가지 매개변수와 일일이 대응하는 네 가지 구성 방법을 제공합니다.이러한 매개변수의 용도와 사용 방법은 ScheduledFutureTask가 사용하는 방법에 따라 달라집니다. ScheduledFutureTask에는 compareTo() 방법이 있습니다.

    public int compareTo(Delayed other) {
        if (other == this) // compare zero if same object
            return 0;
        if (other instanceof ScheduledFutureTask) {
            ScheduledFutureTask> x = (ScheduledFutureTask>)other;
            long diff = time - x.time;
            if (diff < 0)
                return -1;
            else if (diff > 0)
                return 1;
            else if (sequenceNumber < x.sequenceNumber)
                return -1;
            else
                return 1;
        }
        long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
        return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
    }

여러분은 이 방법이 왜 사용되는지 알고 정렬 알고리즘을 제공할 것입니다. 이 알고리즘 규칙은 먼저 타임에 따라 정렬하고 타임은 작은 것이 앞에 있고 큰 것이 뒤에 있으며 타임이 같으면 sequenceNumber로 정렬하고 작은 것이 앞에 있고 큰 것이 뒤에 있습니다.그렇다면 왜 이 클래스에서 compareTo () 방법을 제공합니까?앞에서 설명했듯이 Scheduled Thread Pool Executor가 구조 방법에서 제공하는 것은Delayed WorkQueue () 대기열이다. 즉, Scheduled Thread Pool Executor는 임무를 Delayed WorkQueueue에 추가한 것이고 Delayed WorkQueue는 Delay Queue와 유사하며 내부는 시간을 선후 순서로 하는 대기열을 보호하고 있다.그래서 compareTo () 방법은 DelayedWorkQueue 대기열과 원소ScheduledThreadPoolExecutor task를 정렬하는 알고리즘을 사용합니다.정렬이 해결되었습니다. ScheduledThreadPoolExecutor는task 작업을 어떻게 스케줄링하고 지연시킵니까?모든 스레드의 실행은 run () 방법을 통해 이루어지며, ScheduledThreadPoolExecutor의 run () 방법은 다음과 같습니다.

        public void run() {
            boolean periodic = isPeriodic();
            if (!canRunInCurrentRunState(periodic))
                cancel(false);
            else if (!periodic)
                ScheduledFutureTask.super.run();
            else if (ScheduledFutureTask.super.runAndReset()) {
                setNextRunTime();
                reExecutePeriodic(outerTask);
            }
        }

isPeriodic () 를 호출하여 이 라인이 주기적인 작업 표지인지 확인하고 canRunInCurrentRunState () 방법을 호출하여 이 라인이 실행될 수 있는지 판단하고, 실행할 수 없으면 cancel () 을 호출하여 작업을 취소합니다.

라인이 실행점에 도달했을 때,run () 방법으로task를 실행합니다. 이run () 방법은FutureTask에서 정의된 것입니다.

그렇지 않으면runAndReset () 방법으로 실행하고 충전하며 setNextRunTime () 방법으로 작업의 다음 실행 시간을 계산하고 다시 대기열에 추가하여 이 작업을 반복할 수 있도록 합니다.

isPeriodic() 이 메서드는 지정된 임무가 반복 임무인지 여부를 판단하는 데 사용됩니다.

        public boolean isPeriodic() {
            return period != 0;
        }

canRun InCurrent RunState () 는 임무를 취소할 수 있는지 판단하고, cancel () 은 임무를 취소합니다. 이 두 가지 방법은 비교적 간단합니다.run () 은 임무를 수행하고rund Reset () 는 상태를 실행하며 초기화합니다. 관련이 비교적 넓습니다.Future Task 뒤에 소개합니다.그래서 setNextRunTime()과reExecutePeriodic() 두 가지 지연과 관련된 방법을 중점적으로 소개한다.setNextRunTime () setNextRunTime () 방법은 작업의 다음 실행 시간을 다시 계산하는 데 사용됩니다.다음과 같습니다.

        private void setNextRunTime() {
            long p = period;
            if (p > 0)
                time += p;
            else
                time = triggerTime(-p);
        }

이 방법의 정의는 매우 간단합니다. p > 0,time + = p. 그렇지 않으면 triggerTime () 방법을 사용해서 time를 다시 계산합니다.

    long triggerTime(long delay) {
        return now() +
            ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
    }

reExecutePeriodic

    void reExecutePeriodic(RunnableScheduledFuture> task) {
        if (canRunInCurrentRunState(true)) {
            super.getQueue().add(task);
            if (!canRunInCurrentRunState(true) && remove(task))
                task.cancel(false);
            else
                ensurePrestart();
        }
    }

reExecutePeriodic은 슈퍼를 호출하는 것이 중요하다.getQueue().add(task);작업 task가 포함된 대기열 DelayedWorkQueue에 포함됩니다.ensurePrestart () 는 [Java 덤핑 및 발송] ---J.U.C의 스레드 탱크:ThreadPoolExecutor에서 상세하게 소개했습니다.여기까지ScheduledFutureTask는 이미 소개했고ScheduledFutureTask가ScheduledThreadPoolExecutor에서 역할을 하는 중요성은 말하지 않아도 안다.사실ScheduledThreadPoolExecutor의 실현은 그리 복잡하지 않다. 왜냐하면FutureTask와ThreadPoolExecutor의 지지가 있기 때문에 그 실현은 그렇게 어렵지 않다.