michaelandrepearce edited a comment on issue #2524: ARTEMIS-2240 ActiveMQThreadPoolExecutor should use LinkedTransferQueue URL: https://github.com/apache/activemq-artemis/pull/2524#issuecomment-458039733 @franz1981 a slight improvement, i spotted old master wasn't honoring the pre 2.4 original behaviour of on shutdown, rejection would then use default policy of the executor (AbortPolicy) which would throw rejection exception. a simple correction. --> ``` /* * ActiveMQThreadPoolExecutor: a special ThreadPoolExecutor that combines * the benefits of a cached executor and a fixed size executor. * Similar to a cached executor, threads exceeding the core size are only created on demand, * and will be removed after idling for a specified keep time. * But in contrast to a standard cached executor, tasks are queued if the * maximum pool size if reached, instead of rejected. */ public class ActiveMQThreadPoolExecutor extends ThreadPoolExecutor { /** * The default rejected execution handler */ private static final RejectedExecutionHandler defaultHandler = new AbortPolicy(); // Handler executed when a task is submitted and a new thread cannot be created (because maxSize was reached) // It queues the task on the executors's queue (using the add() method, see ThreadPoolQueue class below) private static class QueueExecutionHandler implements RejectedExecutionHandler { private final RejectedExecutionHandler handler; private QueueExecutionHandler(RejectedExecutionHandler handler) { if (handler == null) throw new NullPointerException(); this.handler = handler; } @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { if (executor.isShutdown() || !executor.getQueue().add(r)) { handler.rejectedExecution(r, executor); } } } // A specialized LinkedBlockingQueue that takes new elements by calling add() but not offer() // This is to force the ThreadPoolExecutor to always create new threads and never queue private static class ThreadPoolQueue extends LinkedTransferQueue<Runnable> { @Override public boolean offer(Runnable runnable) { return tryTransfer(runnable); } @Override public boolean add(Runnable runnable) { return super.offer( runnable ); } } public ActiveMQThreadPoolExecutor(int coreSize, int maxSize, long keep, TimeUnit keepUnits, ThreadFactory factory) { this(coreSize, maxSize, keep, keepUnits, factory, defaultHandler); } public ActiveMQThreadPoolExecutor(int coreSize, int maxSize, long keep, TimeUnit keepUnits, ThreadFactory factory, RejectedExecutionHandler handler) { super(coreSize, maxSize, keep, keepUnits, new ThreadPoolQueue(), factory, new QueueExecutionHandler(handler)); } } ```
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
