java.util.concurrent.BlockingDeque#add ( )源码实例Demo

public static void appendLog(String clusterName, String message) {
    BlockingDeque<String> logQueue = getLogDeque(clusterName);
    if (!Strings.isNullOrEmpty(message)) {
        logQueue.add("[Installation] " + message);
    }
}
 

@Test
public void testImmediateResizing() throws Exception {
    // timing may be an issue here, so we'll use 3 rounds of 1000ms jobs and 700ms pause intervals to give us a
    // buffer of execution time between threads.. this way, local execution time can be between 0 and 300ms
    long start = System.currentTimeMillis();

    // total jobs
    BlockingDeque<Runnable> workDeque = new LinkedBlockingDeque<>();
    for (int i = 0; i < TASK_COUNT; i++) {
        workDeque.add(new LazyJob());
    }

    // first interval
    EnhancedThreadPoolExecutor executor = new EnhancedThreadPoolExecutor(THREAD_COUNT[0], workDeque);
    Assert.assertEquals(THREAD_COUNT[0], executor.prestartAllCoreThreads());

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // verify jobs are removed and active threads is correct
    log.warn("tim: {}, active count: {}", System.currentTimeMillis() - start, executor.getActiveCount());
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0], workDeque.size());
    Assert.assertTrue(Math.abs(executor.getActiveCount() - THREAD_COUNT[0]) < 1);

    // resize pool
    executor.resizeThreadPool(THREAD_COUNT[1]);

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // verify jobs are removed and active threads is correct
    log.warn("tim: {}, active count: {}", System.currentTimeMillis() - start, executor.getActiveCount());
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0] - THREAD_COUNT[1], workDeque.size());
    Assert.assertTrue(Math.abs(executor.getActiveCount() - THREAD_COUNT[1]) < 1);

    // resize pool
    executor.resizeThreadPool(THREAD_COUNT[2]);

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // verify jobs are removed and active threads is correct
    log.warn("tim: {}, active count: {}", System.currentTimeMillis() - start, executor.getActiveCount());
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0] - THREAD_COUNT[1] - THREAD_COUNT[2], workDeque.size());
    Assert.assertTrue(Math.abs(executor.getActiveCount() - THREAD_COUNT[2]) < 1);

    // resize pool
    executor.resizeThreadPool(THREAD_COUNT[3]);

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // verify jobs are removed and active threads is correct
    log.warn("tim: {}, active count: {}", System.currentTimeMillis() - start, executor.getActiveCount());
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0] - THREAD_COUNT[1] - THREAD_COUNT[2] - THREAD_COUNT[3], workDeque.size());
    Assert.assertTrue(Math.abs(executor.getActiveCount() - THREAD_COUNT[3]) < 1);

    executor.shutdown();
}
 

@Test
public void testPauseResume() throws Exception {
    // total jobs
    BlockingDeque<Runnable> workDeque = new LinkedBlockingDeque<>();
    for (int i = 0; i < TASK_COUNT; i++) {
        workDeque.add(new LazyJob());
    }

    // first interval
    EnhancedThreadPoolExecutor executor = new EnhancedThreadPoolExecutor(THREAD_COUNT[0], workDeque);
    Assert.assertEquals(THREAD_COUNT[0], executor.prestartAllCoreThreads());

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // verify jobs are removed and active threads is correct
    log.warn("active count: {}", executor.getActiveCount());
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0], workDeque.size());

    // pause
    Assert.assertTrue(executor.pause());

    // 2nd call should return false, but is idempotent otherwise
    Assert.assertFalse(executor.pause());

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // one round of jobs should be pulled from the queue (they are all waiting to be executed though)
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0] - THREAD_COUNT[0], workDeque.size());

    // resume
    executor.resume();

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // no new jobs should have been pulled from the queue (they were waiting to be executed last interval)
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0] - THREAD_COUNT[0], workDeque.size());

    // wait interval
    Thread.sleep(INTERVAL_TIME);

    // one round of jobs should have been pulled from the queue
    Assert.assertEquals(TASK_COUNT - THREAD_COUNT[0] - THREAD_COUNT[0] - THREAD_COUNT[0], workDeque.size());

    executor.shutdown();
}