[java] ThreadPoolExecutor가 대기열에 들어가기 전에 스레드를 최대로 늘리는 방법은 무엇입니까?

ThreadPoolExecutor더 많은 스레드가 시작되기 전에 대기열 이 제한 되고 가득 차야하는 경우이 제한을 어떻게 해결할 수 있습니까 ?

나는 마침내이 제한에 대한 다소 우아한 (아마도 약간 해키) 해결책을 찾았다 고 생각 ThreadPoolExecutor합니다. 그것은 LinkedBlockingQueue그것을 반환하도록 확장 하는 것을 포함 false합니다queue.offer(...)이미 일부 작업이 대기열 있을 때 . 현재 스레드가 대기중인 작업을 따라 가지 못하는 경우 TPE는 추가 스레드를 추가합니다. 풀이 이미 최대 스레드에있는 경우이 RejectedExecutionHandler호출됩니다. 그런 다음 put(...)대기열로 작업 을 수행하는 핸들러입니다 .

offer(...)반환 할 수 있는 대기열을 작성하는 것은 확실히 이상합니다.false 있고 put()차단하지 않는 . 그러나 이것은 TPE의 대기열 사용과 잘 작동하므로 이것을 수행하는 데 아무런 문제가 없습니다.

코드는 다음과 같습니다.

// extend LinkedBlockingQueue to force offer() to return false conditionally
BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    private static final long serialVersionUID = -6903933921423432194L;
    @Override
    public boolean offer(Runnable e) {
        // Offer it to the queue if there is 0 items already queued, else
        // return false so the TPE will add another thread. If we return false
        // and max threads have been reached then the RejectedExecutionHandler
        // will be called which will do the put into the queue.
        if (size() == 0) {
            return super.offer(e);
        } else {
            return false;
        }
    }
};
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1 /*core*/, 50 /*max*/,
        60 /*secs*/, TimeUnit.SECONDS, queue);
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try {
            // This does the actual put into the queue. Once the max threads
            //  have been reached, the tasks will then queue up.
            executor.getQueue().put(r);
            // we do this after the put() to stop race conditions
            if (executor.isShutdown()) {
                throw new RejectedExecutionException(
                    "Task " + r + " rejected from " + e);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
    }
});

이 메커니즘을 사용하면 작업을 대기열에 제출할 때 ThreadPoolExecutor 됩니다.

1. 스레드 수를 초기에 코어 크기까지 확장합니다 (여기서는 1).
2. 대기열에 제공하십시오. 대기열이 비어 있으면 기존 스레드에서 처리하도록 대기열에 추가됩니다.
3. 큐에 이미 하나 이상의 요소가있는 경우 offer(...)는 false를 반환합니다.
4. false가 반환되면 최대 수 (여기서는 50)에 도달 할 때까지 풀의 스레드 수를 늘립니다.
5. 최대에 있으면 RejectedExecutionHandler
6. 그런 RejectedExecutionHandler다음 FIFO 순서로 사용 가능한 첫 번째 스레드에서 처리 할 작업을 대기열에 넣습니다.

위의 예제 코드에서 큐는 제한되지 않지만 바인딩 된 큐로 정의 할 수도 있습니다. 예를 들어, 1000의 용량을 추가하면 LinkedBlockingQueue다음이 수행됩니다.

1. 스레드를 최대로 확장
2. 1000 개의 작업으로 가득 찰 때까지 대기
3. 그런 다음 대기열에서 공간을 사용할 수있을 때까지 발신자를 차단합니다.

당신이 사용하는 데 필요한 경우에도 offer(...)에서
RejectedExecutionHandler다음은 사용할 수있는 offer(E, long, TimeUnit)대신과 방법을 Long.MAX_VALUE시간 제한 등.

경고:

executor 가 종료 된 후 작업이 추가 될 것으로 예상한다면 executor-service가 종료되었을 때 RejectedExecutionException사용자 지정을 버리는 것이 더 현명 할 수 있습니다 RejectedExecutionHandler. 이것을 지적한 @RaduToader에게 감사드립니다.

편집하다:

이 답변에 대한 또 다른 조정은 유휴 스레드가 있는지 TPE에 요청하고있는 경우에만 항목을 대기열에 넣는 것입니다. 이것에 대한 진정한 클래스를 만들고 ourQueue.setThreadPoolExecutor(tpe);그것에 메소드를 추가 해야합니다.

그러면 offer(...)방법은 다음과 같습니다.

1. tpe.getPoolSize() == tpe.getMaximumPoolSize()이 경우 전화 만하는지 확인하십시오 super.offer(...).
2. 그렇지 않으면 tpe.getPoolSize() > tpe.getActiveCount()전화super.offer(...) 유휴 스레드가있는 것처럼 보이므로 .
3. 그렇지 않으면 false다른 스레드를 포크로 돌아갑니다 .

아마도 이것은 :

int poolSize = tpe.getPoolSize();
int maximumPoolSize = tpe.getMaximumPoolSize();
if (poolSize >= maximumPoolSize || poolSize > tpe.getActiveCount()) {
    return super.offer(e);
} else {
    return false;
}

TPE의 get 메서드는 volatile필드에 액세스 하거나 (의 경우 getActiveCount()) TPE를 잠그고 스레드 목록을 확인하기 때문에 비용이 많이 듭니다 . 또한 여기에는 작업이 부적절하게 대기열에 추가되거나 유휴 스레드가있을 때 다른 스레드가 분기 될 수있는 경합 조건이 있습니다.

답변

    BlockingQueue<Runnable> queue = new LinkedTransferQueue<Runnable>() {
        @Override
        public boolean offer(Runnable e) {
            return tryTransfer(e);
        }
    };
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 50, 60, TimeUnit.SECONDS, queue);
    threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    });

public class GrowBeforeQueueThreadPoolExecutor extends ThreadPoolExecutor {
    private int userSpecifiedCorePoolSize;
    private int taskCount;

    public GrowBeforeQueueThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        userSpecifiedCorePoolSize = corePoolSize;
    }

    @Override
    public void execute(Runnable runnable) {
        synchronized (this) {
            taskCount++;
            setCorePoolSizeToTaskCountWithinBounds();
        }
        super.execute(runnable);
    }

    @Override
    protected void afterExecute(Runnable runnable, Throwable throwable) {
        super.afterExecute(runnable, throwable);
        synchronized (this) {
            taskCount--;
            setCorePoolSizeToTaskCountWithinBounds();
        }
    }

    private void setCorePoolSizeToTaskCountWithinBounds() {
        int threads = taskCount;
        if (threads < userSpecifiedCorePoolSize) threads = userSpecifiedCorePoolSize;
        if (threads > getMaximumPoolSize()) threads = getMaximumPoolSize();
        setCorePoolSize(threads);
    }
}

public void execute(Runnable command) {
    super.execute(command);
    final int poolSize = getPoolSize();
    if (poolSize < getMaximumPoolSize()) {
        if (getQueue().size() > creationThreshold) {
            synchronized (this) {
                setCorePoolSize(poolSize + 1);
                setCorePoolSize(poolSize);
            }
        }
    }
}

final Runnable SENTINEL_NO_OP = new Runnable() { public void run() { } };

final AtomicInteger waitingThreads = new AtomicInteger(0);

BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    @Override
    public boolean offer(Runnable e) {
        // offer returning false will cause the executor to spawn a new thread
        if (e == SENTINEL_NO_OP) return size() <= waitingThreads.get();
        else return super.offer(e);
    }

    @Override
    public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            waitingThreads.incrementAndGet();
            return super.poll(timeout, unit);
        } finally {
            waitingThreads.decrementAndGet();
        }
    }

    @Override
    public Runnable take() throws InterruptedException {
        try {
            waitingThreads.incrementAndGet();
            return super.take();
        } finally {
            waitingThreads.decrementAndGet();
        }
    }
};

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 50, 60, TimeUnit.SECONDS, queue) {
    @Override
    public void execute(Runnable command) {
        super.execute(command);
        if (getQueue().size() > waitingThreads.get()) super.execute(SENTINEL_NO_OP);
    }
};
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (r == SENTINEL_NO_OP) return;
        else throw new RejectedExecutionException();
    }
});