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

/**
 * get of element of invokeAll(c) throws exception on failed task
 */
public void testTimedInvokeAll4() throws Exception {
    final ExecutorService e = new ScheduledThreadPoolExecutor(2);
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new NPETask());
        List<Future<String>> futures =
            e.invokeAll(l, LONG_DELAY_MS, MILLISECONDS);
        assertEquals(1, futures.size());
        try {
            futures.get(0).get();
            shouldThrow();
        } catch (ExecutionException success) {
            assertTrue(success.getCause() instanceof NullPointerException);
        }
    }
}

private RuntimeUtil() {
    scheduler = new ScheduledThreadPoolExecutor(
        schedulerThreads,
        new ThreadFactory() {
            private final AtomicInteger count = new AtomicInteger(0);
            public Thread newThread(Runnable runnable) {
                try {
                    return AccessController.doPrivileged(
                        new NewThreadAction(runnable,
                            "Scheduler(" + count.getAndIncrement() + ")",
                            true));
                } catch (Throwable t) {
                    runtimeLog.log(Level.WARNING,
                                   "scheduler thread factory throws", t);
                    return null;
                }
            }
        });
    /*
     * We would like to allow the scheduler's threads to terminate
     * if possible, but a bug in DelayQueue.poll can cause code
     * like this to result in a busy loop:
     */
    // stpe.setKeepAliveTime(10, TimeUnit.MINUTES);
    // stpe.allowCoreThreadTimeOut(true);
}

public void start() {
    bootstrap.group(group).channel(NioSocketChannel.class);
    bootstrap.option(ChannelOption.TCP_NODELAY, true);
    bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            //                ch.pipeline().addLast(new IdleStateHandler(1, 1, 5));
            ch.pipeline().addLast(new KyroMsgDecoder());
            ch.pipeline().addLast(new KyroMsgEncoder());
            ch.pipeline().addLast(new ClientHandler());
        }
    });

    new ScheduledThreadPoolExecutor(1).scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            scanResponseTable(3000);
        }
    }, 1000, 1000, TimeUnit.MILLISECONDS);
}

public static void main(String[] args) throws Exception {
  final Configuration conf = HBaseConfiguration.create();
  final ChoreService choreService = new ChoreService("CANARY_TOOL");
  final ScheduledChore authChore = AuthUtil.getAuthChore(conf);
  if (authChore != null) {
    choreService.scheduleChore(authChore);
  }

  // loading the generic options to conf
  new GenericOptionsParser(conf, args);

  int numThreads = conf.getInt("hbase.canary.threads.num", MAX_THREADS_NUM);
  LOG.info("Number of exection threads " + numThreads);

  ExecutorService executor = new ScheduledThreadPoolExecutor(numThreads);

  Class<? extends Sink> sinkClass =
      conf.getClass("hbase.canary.sink.class", RegionServerStdOutSink.class, Sink.class);
  Sink sink = ReflectionUtils.newInstance(sinkClass);

  int exitCode = ToolRunner.run(conf, new Canary(executor, sink), args);
  choreService.shutdown();
  executor.shutdown();
  System.exit(exitCode);
}

public static void main(String[] args)
{
    ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
            10);
    Runnable event = new Runnable()
    {

        @Override
        public void run()
        {
            System.out.println("吃饭，睡觉，打豆豆");

        }
    };
    scheduler.schedule(event, 1, TimeUnit.SECONDS);
    scheduler.scheduleAtFixedRate(event, 5, 1, TimeUnit.SECONDS);
}

@PostConstruct
    public void init() {

        scheduledExecutorService = new ScheduledThreadPoolExecutor(1,
                new BasicThreadFactory.Builder().namingPattern("SendNodeServerInfo-schedule-pool-%d").daemon(true).build());
        scheduledExecutorService.scheduleAtFixedRate(() ->
                {
                    //将负载加载到ZK中
                    if (!CollectionUtils.isEmpty(dataCenterChannelStore.getAllChannels())) {
                        dataCenterChannelStore.getAllChannels().stream().forEach(e -> {
                            log.info("channel id:{}, {}", e.id(), e);
                        });
                    }
                    applicationEventPublisher.publishEvent(
                            NodeServerInfoEvent.builder()
                                    .name(goPushNodeServerConfig.getName())
                                    .nodeServerInfo(watch())
                                    .build());
//                写入zk 其实不需要发送 NodeInfoReq
                    nodeSender.send(NodeInfoReq.builder().build());
                }
                , delay, delay, TimeUnit.MILLISECONDS);

    }

public void start() {
    if (ses != null && !ses.isShutdown() && !ses.isTerminated()) {
        return;
    }
    ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(4, new ThreadFactory() {
        private final AtomicInteger threadNumber = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("TimerManager-Worker-" + threadNumber.getAndIncrement());
            return t;
        }
    });
    //this is a no-no, it actually does nothing..then why the fuck are you doing it?
    stpe.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
    ses = stpe;
}

public void testListeningDecorator_scheduleSuccess() throws Exception {
  final CountDownLatch completed = new CountDownLatch(1);
  ScheduledThreadPoolExecutor delegate = new ScheduledThreadPoolExecutor(1) {
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
      completed.countDown();
    }
  };
  ListeningScheduledExecutorService service = listeningDecorator(delegate);
  ListenableFuture<Integer> future =
      service.schedule(Callables.returning(42), 1, TimeUnit.MILLISECONDS);

  /*
   * Wait not just until the Future's value is set (as in future.get()) but
   * also until ListeningScheduledExecutorService's wrapper task is done
   * executing listeners, as detected by yielding control to afterExecute.
   */
  completed.await();
  assertTrue(future.isDone());
  assertThat(future.get()).isEqualTo(42);
  assertListenerRunImmediately(future);
  assertEquals(0, delegate.getQueue().size());
}

public void afterPropertiesSet() throws Exception {
    scheduler = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("Otter-Statistics-Table"),
                                                new ThreadPoolExecutor.CallerRunsPolicy());
    if (statUnit > 0) {
        scheduler.scheduleAtFixedRate(new Runnable() {

            public void run() {
                try {
                    flushBehaviorHistory();
                } catch (Exception e) {
                    logger.error("flush delay stat failed!", e);
                }
            }
        }, statUnit, statUnit, TimeUnit.MILLISECONDS);
    }
}

private RuntimeUtil() {
    scheduler = new ScheduledThreadPoolExecutor(
        schedulerThreads,
        new ThreadFactory() {
            private final AtomicInteger count = new AtomicInteger(0);
            public Thread newThread(Runnable runnable) {
                try {
                    return AccessController.doPrivileged(
                        new NewThreadAction(runnable,
                            "Scheduler(" + count.getAndIncrement() + ")",
                            true));
                } catch (Throwable t) {
                    runtimeLog.log(Level.WARNING,
                                   "scheduler thread factory throws", t);
                    return null;
                }
            }
        });
    /*
     * We would like to allow the scheduler's threads to terminate
     * if possible, but a bug in DelayQueue.poll can cause code
     * like this to result in a busy loop:
     */
    // stpe.setKeepAliveTime(10, TimeUnit.MINUTES);
    // stpe.allowCoreThreadTimeOut(true);
}

/**
 * @param coreThreadPoolPrefix Prefix that will be applied to the Thread name of all threads
 *          spawned by this service
 * @param corePoolSize The initial size to set the core pool of the ScheduledThreadPoolExecutor 
 *          to during initialization. The default size is 1, but specifying a larger size may be
 *          beneficial if you know that 1 thread will not be enough.
 */
public ChoreService(final String coreThreadPoolPrefix, int corePoolSize, boolean jitter) {
  this.coreThreadPoolPrefix = coreThreadPoolPrefix;
  if (corePoolSize < MIN_CORE_POOL_SIZE)  {
    corePoolSize = MIN_CORE_POOL_SIZE;
  }

  final ThreadFactory threadFactory = new ChoreServiceThreadFactory(coreThreadPoolPrefix);
  if (jitter) {
    scheduler = new JitterScheduledThreadPoolExecutorImpl(corePoolSize, threadFactory, 0.1);
  } else {
    scheduler = new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
  }

  scheduler.setRemoveOnCancelPolicy(true);
  scheduledChores = new HashMap<ScheduledChore, ScheduledFuture<?>>();
  choresMissingStartTime = new HashMap<ScheduledChore, Boolean>();
}

/**
 * purge eventually removes cancelled tasks from the queue
 */
public void testPurge() throws InterruptedException {
    final ScheduledFuture[] tasks = new ScheduledFuture[5];
    final Runnable releaser = new Runnable() { public void run() {
        for (ScheduledFuture task : tasks)
            if (task != null) task.cancel(true); }};
    final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1);
    try (PoolCleaner cleaner = cleaner(p, releaser)) {
        for (int i = 0; i < tasks.length; i++)
            tasks[i] = p.schedule(new SmallPossiblyInterruptedRunnable(),
                                  LONG_DELAY_MS, MILLISECONDS);
        int max = tasks.length;
        if (tasks[4].cancel(true)) --max;
        if (tasks[3].cancel(true)) --max;
        // There must eventually be an interference-free point at
        // which purge will not fail. (At worst, when queue is empty.)
        long startTime = System.nanoTime();
        do {
            p.purge();
            long count = p.getTaskCount();
            if (count == max)
                return;
        } while (millisElapsedSince(startTime) < LONG_DELAY_MS);
        fail("Purge failed to remove cancelled tasks");
    }
}

/**
 * scheduleAtFixedRate executes series of tasks at given rate.
 * Eventually, it must hold that:
 *   cycles - 1 <= elapsedMillis/delay < cycles
 */
public void testFixedRateSequence() throws InterruptedException {
    final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1);
    try (PoolCleaner cleaner = cleaner(p)) {
        for (int delay = 1; delay <= LONG_DELAY_MS; delay *= 3) {
            final long startTime = System.nanoTime();
            final int cycles = 8;
            final CountDownLatch done = new CountDownLatch(cycles);
            final Runnable task = new CheckedRunnable() {
                public void realRun() { done.countDown(); }};
            final ScheduledFuture periodicTask =
                p.scheduleAtFixedRate(task, 0, delay, MILLISECONDS);
            final int totalDelayMillis = (cycles - 1) * delay;
            await(done, totalDelayMillis + LONG_DELAY_MS);
            periodicTask.cancel(true);
            final long elapsedMillis = millisElapsedSince(startTime);
            assertTrue(elapsedMillis >= totalDelayMillis);
            if (elapsedMillis <= cycles * delay)
                return;
            // else retry with longer delay
        }
        fail("unexpected execution rate");
    }
}

private ChatSDKAbstractConversationsFragmentChatSDKThreadPool(){

    if (NUMBER_OF_CORES <= 0)
        NUMBER_OF_CORES = 2;

    // Creates a thread pool manager
    threadPool = new ThreadPoolExecutor(
            NUMBER_OF_CORES,       // Initial pool size
            NUMBER_OF_CORES,       // Max pool size
            KEEP_ALIVE_TIME,
            KEEP_ALIVE_TIME_UNIT,
            workQueue);

    scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(NUMBER_OF_CORES);

}

@Override
protected ScheduledExecutorService scheduler() {
  if (this.scheduler == null) {
    synchronized (this) {
      if (this.scheduler == null) {
        ThreadFactory timerFactory = new ThreadFactoryBuilder()
            .setNameFormat("AsyncReporter-" + id + "-timer-%d")
            .setDaemon(true)
            .build();
        ScheduledThreadPoolExecutor timerPool = new ScheduledThreadPoolExecutor(timerThreads, timerFactory);
        timerPool.setRemoveOnCancelPolicy(true);
        this.scheduler = timerPool;
        return timerPool;
      }
    }
  }
  return scheduler;
}

void scheduleAtTheEndOfTime(ScheduledThreadPoolExecutor pool,
                            Runnable r, int how) {
    switch (how) {
    case 0:
        pool.schedule(r, Long.MAX_VALUE, MILLISECONDS);
        break;
    case 1:
        pool.schedule(Executors.callable(r), Long.MAX_VALUE, DAYS);
        break;
    case 2:
        pool.scheduleWithFixedDelay(r, Long.MAX_VALUE, 1000, NANOSECONDS);
        break;
    case 3:
        pool.scheduleAtFixedRate(r, Long.MAX_VALUE, 1000, MILLISECONDS);
        break;
    default:
        fail(String.valueOf(how));
    }
}

@Test
public void clientExecutionTimeoutEnabled_RequestCompletesWithinTimeout_TaskCanceled() throws Exception {
    AmazonHttpClient httpClient = HttpTestUtils.testClientBuilder()
                                               .httpClient(sdkHttpClient)
                                               .retryPolicy(RetryPolicy.NONE)
                                               .build();

    try {
        ClientExecutionAndRequestTimerTestUtils
                .execute(httpClient, ClientExecutionAndRequestTimerTestUtils.createMockGetRequest());
        fail("Exception expected");
    } catch (SdkClientException e) {
        NullResponseHandler.assertIsUnmarshallingException(e);
    }

    ScheduledThreadPoolExecutor requestTimerExecutor = httpClient.getClientExecutionTimer().getExecutor();
    ClientExecutionAndRequestTimerTestUtils.assertTimerNeverTriggered(requestTimerExecutor);
    ClientExecutionAndRequestTimerTestUtils.assertCanceledTasksRemoved(requestTimerExecutor);
    // Core threads should be spun up on demand. Since only one task was submitted only one
    // thread should exist
    assertEquals(1, requestTimerExecutor.getPoolSize());
    ClientExecutionAndRequestTimerTestUtils.assertCoreThreadsShutDownAfterBeingIdle(requestTimerExecutor);
}

/**
 * scheduleAtFixedRate executes runnable after given initial delay
 */
public void testSchedule4() throws Exception {
    final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1);
    try (PoolCleaner cleaner = cleaner(p)) {
        final long startTime = System.nanoTime();
        final CountDownLatch done = new CountDownLatch(1);
        Runnable task = new CheckedRunnable() {
            public void realRun() {
                done.countDown();
                assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
            }};
        ScheduledFuture f =
            p.scheduleAtFixedRate(task, timeoutMillis(),
                                  LONG_DELAY_MS, MILLISECONDS);
        await(done);
        assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
        f.cancel(true);
    }
}

public ServerConnector(int ownPort, String host, int port) {
    if (ownPort == port) {
        System.out.println("Cannot connect to own server!");
        return;
    }

    server = new EchoServer(ownPort);
    client = new EchoClient(host, port);

    ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);

    executor.execute(() -> {
        try {
            server.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
}

static void purgeExecutors() {
    try {
        Iterator<ScheduledThreadPoolExecutor> it = EXECUTORS.keySet().iterator();
        while (it.hasNext()) {
            ScheduledThreadPoolExecutor exec = (ScheduledThreadPoolExecutor) it.next();
            if (exec.isShutdown()) {
                it.remove();
            } else {
                exec.purge();
            }
        }
    } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        RxJavaPlugins.getInstance().getErrorHandler().handleError(t);
    }
}

public ResourceLocalizationService(Dispatcher dispatcher,
    ContainerExecutor exec, DeletionService delService,
    LocalDirsHandlerService dirsHandler, Context context) {

  super(ResourceLocalizationService.class.getName());
  this.exec = exec;
  this.dispatcher = dispatcher;
  this.delService = delService;
  this.dirsHandler = dirsHandler;

  this.cacheCleanup = new ScheduledThreadPoolExecutor(1,
      new ThreadFactoryBuilder()
        .setNameFormat("ResourceLocalizationService Cache Cleanup")
        .build());
  this.stateStore = context.getNMStateStore();
  this.nmContext = context;
}

public void runTest()
{
    ScheduledExecutorService s=new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("Scheduled"));
    s.scheduleAtFixedRate(map.getCleanTask(),1, 1, TimeUnit.SECONDS);
    ExecutorService es=Executors.newCachedThreadPool();
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    s.scheduleAtFixedRate(this::printInfo,5, 5, TimeUnit.SECONDS);
}

public void runTest()
{
    ScheduledExecutorService s=new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("Scheduled"));
    s.scheduleAtFixedRate(map.getCleanTask(),1, 1, TimeUnit.SECONDS);
    ExecutorService es=Executors.newCachedThreadPool();
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::writeTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    es.execute(this::readTest);
    s.scheduleAtFixedRate(this::printInfo,5, 5, TimeUnit.SECONDS);
}

private TxManagerLocator() {
    List<TxManagerServiceDTO> initial = Lists.newArrayList();
    mConfigservices = new AtomicReference<>(initial);
    mResponsetype = new TypeToken<List<TxManagerServiceDTO>>() {
    }.getType();
    this.mExecutorservice = new ScheduledThreadPoolExecutor(1,
            TxTransactionThreadFactory.create("TxManagerLocator", true));
}

@PostConstruct
public void init() {
    scheduledExecutorService =  new ScheduledThreadPoolExecutor(txConfig.getTransactionThreadMax(),
            TxTransactionThreadFactory.create(THREAD_FACTORY_NAME, false));

    fixExecutorService = new ThreadPoolExecutor(txConfig.getTransactionThreadMax(), txConfig.getTransactionThreadMax(), 0, TimeUnit.MILLISECONDS,
            createBlockingQueue(),
            TxTransactionThreadFactory.create(THREAD_FACTORY_NAME, false), createPolicy());

}

/**
 * getTaskCount increases, but doesn't overestimate, when tasks
 * submitted
 */
public void testGetTaskCount() throws InterruptedException {
    final int TASKS = 3;
    final CountDownLatch done = new CountDownLatch(1);
    final ThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1);
    try (PoolCleaner cleaner = cleaner(p, done)) {
        final CountDownLatch threadStarted = new CountDownLatch(1);
        assertEquals(0, p.getTaskCount());
        assertEquals(0, p.getCompletedTaskCount());
        p.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadStarted.countDown();
                await(done);
            }});
        await(threadStarted);
        assertEquals(1, p.getTaskCount());
        assertEquals(0, p.getCompletedTaskCount());
        for (int i = 0; i < TASKS; i++) {
            assertEquals(1 + i, p.getTaskCount());
            p.execute(new CheckedRunnable() {
                public void realRun() throws InterruptedException {
                    threadStarted.countDown();
                    assertEquals(1 + TASKS, p.getTaskCount());
                    await(done);
                }});
        }
        assertEquals(1 + TASKS, p.getTaskCount());
        assertEquals(0, p.getCompletedTaskCount());
    }
    assertEquals(1 + TASKS, p.getTaskCount());
    assertEquals(1 + TASKS, p.getCompletedTaskCount());
}

/**
 * Submitting null tasks throws NullPointerException
 */
public void testNullTaskSubmission() {
    final ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1);
    try (PoolCleaner cleaner = cleaner(p)) {
        assertNullTaskSubmissionThrowsNullPointerException(p);
    }
}

/**
 * Waits until a little after the thread pools keep alive time and then asserts that all thre
 *
 * @param timerExecutor Executor used by timer implementation
 */
public static void assertCoreThreadsShutDownAfterBeingIdle(ScheduledThreadPoolExecutor timerExecutor) {
    try {
        Thread.sleep(timerExecutor.getKeepAliveTime(TimeUnit.MILLISECONDS) + 1000);
    } catch (InterruptedException ignored) {
        // Ignored.
    }
    assertEquals(0, timerExecutor.getPoolSize());
}

public void afterPropertiesSet() throws Exception {
    scheduler = new ScheduledThreadPoolExecutor(DEFAULT_POOL, new NamedThreadFactory("Otter-Statistics-Client"),
                                                new ThreadPoolExecutor.CallerRunsPolicy());
    scheduler.submit(new Runnable() {

        public void run() {
            doSendDelayCountEvent();
        }
    });
}

/**
 * timed invokeAll(null) throws NPE
 */
public void testTimedInvokeAll1() throws Exception {
    final ExecutorService e = new ScheduledThreadPoolExecutor(2);
    try (PoolCleaner cleaner = cleaner(e)) {
        try {
            e.invokeAll(null, randomTimeout(), randomTimeUnit());
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

/**
 * timed invokeAll(,,null) throws NPE
 */
public void testTimedInvokeAllNullTimeUnit() throws Exception {
    final ExecutorService e = new ScheduledThreadPoolExecutor(2);
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new StringTask());
        try {
            e.invokeAll(l, randomTimeout(), null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

/**
 * Have shutdown actually means shutdown. Tasks that need to complete should use
 * futures.
 */
public static ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor(String name, UncaughtExceptionHandler handler, int poolSize, int stackSize) {
    // HACK: ScheduledThreadPoolExecutor won't let use the handler so
    // if we're using ExceptionHandlingRunnable then we'll be able to 
    // pick up the exceptions
    Thread.setDefaultUncaughtExceptionHandler(handler);

    ThreadFactory factory = getThreadFactory(name, handler);
    ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(poolSize, factory);
    executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
    executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
    return executor;
}

public void addToBlacklistedClient(ClientProxyMembershipID proxyID) {
  blackListedClients.add(proxyID);
  // ensure that cache and distributed system state are current and open
  this.getCache();
  new ScheduledThreadPoolExecutor(1).schedule(new ExpireBlackListTask(proxyID), 120,
      TimeUnit.SECONDS);
}

/**
 * Canary entry point for specified table with task type(read/write)
 * @throws Exception
 */
public static void sniff(final Admin admin, TableName tableName, TaskType taskType)
    throws Exception {
  List<Future<Void>> taskFutures =
      Canary.sniff(admin, new StdOutSink(), tableName.getNameAsString(),
        new ScheduledThreadPoolExecutor(1), taskType);
  for (Future<Void> future : taskFutures) {
    future.get();
  }
}

@Inject
public DataStreamWebsocket(SimulationStore simStore, DataStreamStore dataStore, ObjectMapper mapper, ScheduledThreadPoolExecutor executor) {
    this.simStore = simStore;
    this.dataStore = dataStore;
    this.mapper = mapper;
    this.executor = executor;
}

/**
 * invokeAll(null) throws NPE
 */
public void testInvokeAll1() throws Exception {
    final ExecutorService e = new ScheduledThreadPoolExecutor(2);
    try (PoolCleaner cleaner = cleaner(e)) {
        try {
            e.invokeAll(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

/**
 * timed invokeAll(c) throws NPE if c has null elements
 */
public void testTimedInvokeAll3() throws Exception {
    final ExecutorService e = new ScheduledThreadPoolExecutor(2);
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new StringTask());
        l.add(null);
        try {
            e.invokeAll(l, randomTimeout(), randomTimeUnit());
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

/**
 * completed submit of runnable returns successfully
 */
public void testSubmitRunnable() throws Exception {
    final ExecutorService e = new ScheduledThreadPoolExecutor(2);
    try (PoolCleaner cleaner = cleaner(e)) {
        Future<?> future = e.submit(new NoOpRunnable());
        future.get();
        assertTrue(future.isDone());
    }
}

public NewThreadWorker(ThreadFactory threadFactory) {
    ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory);
    if (!tryEnableCancelPolicy(exec) && (exec instanceof ScheduledThreadPoolExecutor)) {
        registerExecutor((ScheduledThreadPoolExecutor) exec);
    }
    this.schedulersHook = RxJavaPlugins.getInstance().getSchedulersHook();
    this.executor = exec;
}

public void start() throws MQClientException {
    final String group = this.defaultMQPullConsumer.getConsumerGroup();
    this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(
        this.pullThreadNums,
        new ThreadFactoryImpl("PullMsgThread-" + group)
    );

    this.defaultMQPullConsumer.setMessageQueueListener(this.messageQueueListener);

    this.defaultMQPullConsumer.start();

    log.info("MQPullConsumerScheduleService start OK, {} {}",
        this.defaultMQPullConsumer.getConsumerGroup(), this.callbackTable);
}