Java 类java.util.concurrent.RejectedExecutionHandler 实例源码

private RejectedExecutionHandler createPolicy() {
    RejectedPolicyTypeEnum rejectedPolicyType = RejectedPolicyTypeEnum.fromString(txConfig.getRejectPolicy());

    switch (rejectedPolicyType) {
        case BLOCKING_POLICY:
            return new BlockingPolicy();
        case CALLER_RUNS_POLICY:
            return new CallerRunsPolicy();
        case ABORT_POLICY:
            return new AbortPolicy();
        case REJECTED_POLICY:
            return new RejectedPolicy();
        case DISCARDED_POLICY:
            return new DiscardedPolicy();
        default:
            return new RejectedPolicy();
    }
}

private RejectedExecutionHandler createPolicy() {
    RejectedPolicyTypeEnum rejectedPolicyType = RejectedPolicyTypeEnum.fromString(tccConfig.getRejectPolicy());
    switch (rejectedPolicyType) {
        case BLOCKING_POLICY:
            return new BlockingPolicy();
        case CALLER_RUNS_POLICY:
            return new CallerRunsPolicy();
        case ABORT_POLICY:
            return new AbortPolicy();
        case REJECTED_POLICY:
            return new RejectedPolicy();
        case DISCARDED_POLICY:
            return new DiscardedPolicy();
        default:
            return new AbortPolicy();
    }
}

private RejectedExecutionHandler createPolicy() {
    RejectedPolicyTypeEnum rejectedPolicyType = RejectedPolicyTypeEnum.fromString(mythConfig.getRejectPolicy());
    switch (rejectedPolicyType) {
        case BLOCKING_POLICY:
            return new BlockingPolicy();
        case CALLER_RUNS_POLICY:
            return new CallerRunsPolicy();
        case ABORT_POLICY:
            return new AbortPolicy();
        case REJECTED_POLICY:
            return new RejectedPolicy();
        case DISCARDED_POLICY:
            return new DiscardedPolicy();
        default:
            return new DiscardedPolicy();
    }
}

@Override
protected ExecutorService initializeExecutor(
        ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    ScheduledExecutorService executor =
            createExecutor(this.poolSize, threadFactory, rejectedExecutionHandler);

    if (executor instanceof ScheduledThreadPoolExecutor && this.removeOnCancelPolicy != null) {
        ((ScheduledThreadPoolExecutor) executor).setRemoveOnCancelPolicy(this.removeOnCancelPolicy);
    }

    // Register specified ScheduledExecutorTasks, if necessary.
    if (!ObjectUtils.isEmpty(this.scheduledExecutorTasks)) {
        registerTasks(this.scheduledExecutorTasks, executor);
    }

    // Wrap executor with an unconfigurable decorator.
    this.exposedExecutor = (this.exposeUnconfigurableExecutor ?
            Executors.unconfigurableScheduledExecutorService(executor) : executor);

    return executor;
}

@Override
protected ExecutorService initializeExecutor(
        ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
    ThreadPoolExecutor executor  = createExecutor(this.corePoolSize, this.maxPoolSize,
            this.keepAliveSeconds, queue, threadFactory, rejectedExecutionHandler);
    if (this.allowCoreThreadTimeOut) {
        executor.allowCoreThreadTimeOut(true);
    }

    // Wrap executor with an unconfigurable decorator.
    this.exposedExecutor = (this.exposeUnconfigurableExecutor ?
            Executors.unconfigurableExecutorService(executor) : executor);

    return executor;
}

public HadoopThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    RejectedExecutionHandler handler) {
  super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
      handler);
}

public HadoopThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler) {
  super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
      threadFactory, handler);
}

public LifecycleModule() {
    // The thread pool is unbounded, so use direct handoff
    BlockingQueue<Runnable> queue = new SynchronousQueue<Runnable>();
    // Discard tasks that are submitted during shutdown
    RejectedExecutionHandler policy =
            new ThreadPoolExecutor.DiscardPolicy();
    // Create threads as required and keep them in the pool for 60 seconds
    ioExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
            60, SECONDS, queue, policy);
}

public CryptoModule() {
    // Use an unbounded queue
    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
    // Discard tasks that are submitted during shutdown
    RejectedExecutionHandler policy =
            new ThreadPoolExecutor.DiscardPolicy();
    // Create a limited # of threads and keep them in the pool for 60 secs
    cryptoExecutor = new TimeLoggingExecutor("CryptoExecutor", 0,
            MAX_EXECUTOR_THREADS, 60, SECONDS, queue, policy);
}

public DatabaseExecutorModule() {
    // Use an unbounded queue
    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
    // Discard tasks that are submitted during shutdown
    RejectedExecutionHandler policy =
            new ThreadPoolExecutor.DiscardPolicy();
    // Use a single thread and keep it in the pool for 60 secs
    databaseExecutor = new TimeLoggingExecutor("DatabaseExecutor", 0, 1,
            60, SECONDS, queue, policy);
}

public TimeLoggingExecutor(String tag, int corePoolSize, int maxPoolSize,
        long keepAliveTime, TimeUnit unit,
        BlockingQueue<Runnable> workQueue,
        RejectedExecutionHandler handler) {
    super(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue,
            handler);
    log = Logger.getLogger(tag);
}

public LifecycleModule() {
    // The thread pool is unbounded, so use direct handoff
    BlockingQueue<Runnable> queue = new SynchronousQueue<Runnable>();
    // Discard tasks that are submitted during shutdown
    RejectedExecutionHandler policy =
            new ThreadPoolExecutor.DiscardPolicy();
    // Create threads as required and keep them in the pool for 60 seconds
    ioExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
            60, SECONDS, queue, policy);
}

public CryptoModule() {
    // Use an unbounded queue
    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
    // Discard tasks that are submitted during shutdown
    RejectedExecutionHandler policy =
            new ThreadPoolExecutor.DiscardPolicy();
    // Create a limited # of threads and keep them in the pool for 60 secs
    cryptoExecutor = new TimeLoggingExecutor("CryptoExecutor", 0,
            MAX_EXECUTOR_THREADS, 60, SECONDS, queue, policy);
}

protected synchronized static void createPool() {
    BlockingQueue<Runnable> workers = new LinkedBlockingQueue<>();
    RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
    mDownloadPool = new DownloadPool(
            corePoolSize,
            maxPoolSize,
            keepAliveTime,
            TimeUnit.MILLISECONDS,
            workers,
            Executors.defaultThreadFactory(),
            handler);
}

@Override
protected ExecutorService initializeExecutor (ThreadFactory threadFactory,
   RejectedExecutionHandler rejectedExecutionHandler)
{
   FairQueue<Runnable> queue = new FairQueue<Runnable> ();
   ThreadPoolExecutor executor =
      new ThreadPoolExecutor (this.corePoolSize, this.corePoolSize,
         keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory,
         rejectedExecutionHandler);

   this.threadPoolExecutor = executor;
   return executor;
}

protected ThreadPoolExecutor createThreadPool(ThreadGroup parentGroup, String name, int minThreadCound,
        int maxThreadCount, int queueSize, RejectedExecutionHandler rejectPolicy){
    ThreadFactory threadFactory = new NamedThreadFactory(parentGroup, name, true);
    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(queueSize);
    return new ThreadPoolExecutor(minThreadCound, maxThreadCount, 1, TimeUnit.MINUTES, queue, threadFactory,
            rejectPolicy);
}

@Override
protected ExecutorService initializeExecutor(
        ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
    ThreadPoolExecutor executor  = new ThreadPoolExecutor(
            this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
            queue, threadFactory, rejectedExecutionHandler);
    if (this.allowCoreThreadTimeOut) {
        executor.allowCoreThreadTimeOut(true);
    }

    this.threadPoolExecutor = executor;
    return executor;
}

@Override
protected ExecutorService initializeExecutor(
        ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    this.scheduledExecutor = createExecutor(this.poolSize, threadFactory, rejectedExecutionHandler);

    if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor && this.removeOnCancelPolicy != null) {
        ((ScheduledThreadPoolExecutor) this.scheduledExecutor).setRemoveOnCancelPolicy(this.removeOnCancelPolicy);
    }

    return this.scheduledExecutor;
}

public ThreadPoolQueueGroupExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
        BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler,
        IndexQueueGroupManager iqm,KeyQueueGroupManager kqm) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    if (iqm==null || kqm==null)
    {
        throw new IllegalArgumentException();
    }
    this.iqm=iqm;
    this.kqm=kqm;
    init();
}

public ScheduledThreadPoolQueueGroupExecutor(int corePoolSize, ThreadFactory threadFactory,RejectedExecutionHandler handler,
        IndexQueueGroupManager iqm,KeyQueueGroupManager kqm) {
    super(corePoolSize, threadFactory, handler);
    if (iqm==null || kqm==null)
    {
        throw new IllegalArgumentException();
    }
    this.iqm=iqm;
    this.kqm=kqm;
    init();
}

public StandardThreadExecutor(int coreThreads, int maxThreads, long keepAliveTime,
                              TimeUnit unit, int queueCapacity, ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
    super(coreThreads, maxThreads, keepAliveTime, unit, new ExecutorQueue(), threadFactory,
        handler);
    ((ExecutorQueue) getQueue()).setStandardThreadExecutor(this);

    submittedTasksCount = new AtomicInteger(0);

    // 最大并发任务限制： 队列buffer数 + 最大线程数 
    maxSubmittedTaskCount = queueCapacity + maxThreads;
}

/**
 * 得到callback用的线程池
 *
 * @param build
 *         没有时是否构建
 * @return callback用的线程池
 */
public synchronized static ThreadPoolExecutor getCallbackThreadPool(boolean build) {
    if (callbackThreadPool == null && build) {
        // 一些系统参数，可以从配置或者注册中心获取。
        int coresize = CommonUtils.parseInt(JSFContext.getGlobalVal(Constants.SETTING_CALLBACK_POOL_CORE_SIZE, null), Constants.DEFAULT_CLIENT_CALLBACK_CORE_THREADS);
        int maxsize = CommonUtils.parseInt(JSFContext.getGlobalVal(Constants.SETTING_CALLBACK_POOL_MAX_SIZE, null), Constants.DEFAULT_CLIENT_CALLBACK_MAX_THREADS);
        int queuesize = CommonUtils.parseInt(JSFContext.getGlobalVal(Constants.SETTING_CALLBACK_POOL_QUEUE, null), Constants.DEFAULT_CLIENT_CALLBACK_QUEUE);
        BlockingQueue<Runnable> queue = ThreadPoolUtils.buildQueue(queuesize);
        NamedThreadFactory threadFactory = new NamedThreadFactory("JSF-CLI-CB", true);

        RejectedExecutionHandler handler = new RejectedExecutionHandler() {
            private int i = 1;

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (i++ % 7 == 0) {
                    i = 1;
                    logger.warn("Task:{} has been reject for ThreadPool exhausted!" +
                                    " pool:{}, active:{}, queue:{}, taskcnt: {}",
                            new Object[]{
                                    r,
                                    executor.getPoolSize(),
                                    executor.getActiveCount(),
                                    executor.getQueue().size(),
                                    executor.getTaskCount()
                            });
                }
                throw new RejectedExecutionException("Callback handler thread pool has bean exhausted");
            }
        };
        callbackThreadPool = ThreadPoolUtils.newCachedThreadPool(coresize, maxsize, queue, threadFactory, handler);
    }
    return callbackThreadPool;
}

/**
 * getRejectedExecutionHandler returns handler in constructor if not set
 */
public void testGetRejectedExecutionHandler() {
    final RejectedExecutionHandler handler = new NoOpREHandler();
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(1, 2,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10),
                               handler);
    try (PoolCleaner cleaner = cleaner(p)) {
        assertSame(handler, p.getRejectedExecutionHandler());
    }
}

/**
 * setRejectedExecutionHandler sets the handler returned by
 * getRejectedExecutionHandler
 */
public void testSetRejectedExecutionHandler() {
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(1, 2,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        RejectedExecutionHandler handler = new NoOpREHandler();
        p.setRejectedExecutionHandler(handler);
        assertSame(handler, p.getRejectedExecutionHandler());
    }
}

/**
 * Constructor throws if handler is set to null
 */
public void testConstructorNullPointerException5() {
    try {
        new ThreadPoolExecutor(1, 2, 1L, SECONDS,
                               new ArrayBlockingQueue<Runnable>(10),
                               (RejectedExecutionHandler) null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * Constructor throws if handler is null
 */
public void testConstructorNullPointerException7() {
    try {
        new ThreadPoolExecutor(1, 2, 1L, SECONDS,
                               new ArrayBlockingQueue<Runnable>(10),
                               new SimpleThreadFactory(),
                               (RejectedExecutionHandler) null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

CustomTPE(int corePoolSize,
          int maximumPoolSize,
          long keepAliveTime,
          TimeUnit unit,
          BlockingQueue<Runnable> workQueue,
          RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
      handler);
}

CustomTPE(int corePoolSize,
          int maximumPoolSize,
          long keepAliveTime,
          TimeUnit unit,
          BlockingQueue<Runnable> workQueue,
          ThreadFactory threadFactory,
          RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
      workQueue, threadFactory, handler);
}

/**
 * getRejectedExecutionHandler returns handler in constructor if not set
 */
public void testGetRejectedExecutionHandler() {
    final RejectedExecutionHandler handler = new NoOpREHandler();
    final ThreadPoolExecutor p =
        new CustomTPE(1, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10),
                      handler);
    try (PoolCleaner cleaner = cleaner(p)) {
        assertSame(handler, p.getRejectedExecutionHandler());
    }
}

/**
 * setRejectedExecutionHandler sets the handler returned by
 * getRejectedExecutionHandler
 */
public void testSetRejectedExecutionHandler() {
    final ThreadPoolExecutor p =
        new CustomTPE(1, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        RejectedExecutionHandler handler = new NoOpREHandler();
        p.setRejectedExecutionHandler(handler);
        assertSame(handler, p.getRejectedExecutionHandler());
    }
}

/**
 * Constructor throws if handler is set to null
 */
public void testConstructorNullPointerException5() {
    try {
        new CustomTPE(1, 2, 1L, SECONDS,
                      new ArrayBlockingQueue<Runnable>(10),
                      (RejectedExecutionHandler) null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * Constructor throws if handler is null
 */
public void testConstructorNullPointerException7() {
    try {
        new CustomTPE(1, 2, 1L, SECONDS,
                      new ArrayBlockingQueue<Runnable>(10),
                      new SimpleThreadFactory(),
                      (RejectedExecutionHandler) null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

public ThreadPoolTaskExecutor(int corePoolSize, int maximumPoolSize, 
                              long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, 
                              RejectedExecutionHandler handler) {

    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}

public ThreadPoolTaskExecutor(int corePoolSize, int maximumPoolSize, 
                              long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, 
                              ThreadFactory threadFactory, 
                              RejectedExecutionHandler handler) {

    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}

@Override
public Long getRejectedTaskCount() {
    RejectedExecutionHandler rejectedHandler = executor.getRejectedExecutionHandler();
    if(rejectedHandler instanceof CountingRejectedExecutionHandler) {
        return Long.valueOf(((CountingRejectedExecutionHandler)rejectedHandler)
                                                                 .getRejectedTaskCount());
    }

    return null;
}

public XExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}

public XExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}

public HadoopScheduledThreadPoolExecutor(int corePoolSize,
    RejectedExecutionHandler handler) {
  super(corePoolSize, handler);
}

public HadoopScheduledThreadPoolExecutor(int corePoolSize,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler) {
  super(corePoolSize, threadFactory, handler);
}

public BrokerFixedThreadPoolExecutor(final int corePoolSize, final int maximumPoolSize, final long keepAliveTime, final TimeUnit unit,
    final BlockingQueue<Runnable> workQueue, final RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}