activemq-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] michaelandrepearce commented on issue #2524: ARTEMIS-2240 ActiveMQThreadPoolExecutor should use LinkedTransferQueue
Date Mon, 28 Jan 2019 08:38:45 GMT
michaelandrepearce commented 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) {
            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:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message