Some points needing attention about thread pool

Some points needing attention about thread pool

Specify the thread name , exception handling

Customize ThreadFactory Appoint

public static final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), (r) -> {
    
    Thread thread = new Thread(r);
    thread.setName("xxxx-pool-" + atomicInteger.getAndIncrement());
    thread.setUncaughtExceptionHandler((t, e) -> {
    
        loggerHelper.errorLog(t.getName() + "  Thread exception handling  : " + e.getMessage());
        e.printStackTrace();
        System.out.println(t.getName() + "  Thread exception handling  : " + e.getMessage());

    });
    return thread;
}, new ThreadPoolExecutor.AbortPolicy());

ThreadPoolTaskExecutor Prefix specifying method

 AtomicInteger atomicInteger = new AtomicInteger();

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
		//  Prefix specifying method 
        //executor.setThreadNamePrefix("xxxx");

        // rejection-policy： When pool Already achieved max size When , How to deal with new tasks 
        // CALLER_RUNS： Do not execute tasks in New Threads , It's the thread where the caller is running 
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        executor.setThreadFactory((runnable)->{
    
            Thread t = new Thread(runnable);
            t.setName(threadNamePrefix + atomicInteger.getAndIncrement());
            t.setUncaughtExceptionHandler((th, e) -> {
    
                loggerHelper.errorLog(th.getName() + " xxxxxx  Thread exception handling  : " + e.getMessage());
                e.printStackTrace();
            });

            return t;
        });

        return executor;

Thread pool thread default construction

    static class DefaultThreadFactory implements ThreadFactory {
    
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() {
    
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
    
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

Threads Pool queue Type to be discussed

SynchronousQueue queue Applicable scenario ：
A short-term task

characteristic ： A short-term task , No capacity queue , Submit a task and create a thread .

Use SynchronousQueue queue , The characteristic is that there is no capacity , You can't put it in without threads .

For tasks that take longer , It is recommended to use bounded blocking queues ArrayBlockingQueue,LinkedBlockingQueue It is recommended to specify the capacity

Thread pool threads , And for different business thread pools

core The number of cores and the maximum number of cores are set according to the actual business , Specific pressure test is required 

Thread pool shutdown problem ( local variable , Remember to close the thread pool )

Otherwise, the number of threads will increase at a time ,core Will not be recycled

Test examples

    static void test(){
    
        // In the way   Define thread pool 
        ExecutorService pool = new org.apache.tomcat.util.threads.ThreadPoolExecutor(1, 2, 10, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                Executors.defaultThreadFactory(),
                new org.apache.tomcat.util.threads.ThreadPoolExecutor.AbortPolicy());

        pool.execute(()->{
    
            System.out.println(System.currentTimeMillis());
        });
      //  Turn tests on and off 
       // pool.shutdown();

    }

    public static void main(String[] args) throws InterruptedException {
    
        while (true){
    
            TimeUnit.SECONDS.sleep(2);
            test();
        }

    }

Use visualVm You can observe whether the number of process threads is stable

Close thread

pool.shutdown(); // Set interrupt flag , The thread waiting to execute the task is interrupted after execution , Immediately interrupt other idle threads
pool.shutdownNow(); // Break all threads , The task in the task queue will be returned

according to jvm Specification shutdown thread , Close the daemon thread , Non - daemon threads in daemon threads will also be shut down

Examples are as follows

public static final ThreadPoolExecutor cardPool = new ThreadPoolExecutor(1, 4, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(MAX_TASK_SIZE),
            (r) -> {
    
                Thread t = new Thread(r);
                t.setDaemon(true);
                t.setName("xxxx" + ai.getAndIncrement());
                t.setUncaughtExceptionHandler((th,ex)->{
    
                    //System.out.println(th.getName() + " " + ex.getMessage());
                    loggerHelper.errorLog(th.getName() + " " + ex.getMessage());
                    ex.printStackTrace();
                });
                return t;
            });

    public static void main(String[] args) throws InterruptedException {
    
        cardPool.execute(()->{
    
            try {
    
                TimeUnit.SECONDS.sleep(6);
            } catch (InterruptedException e) {
    
                e.printStackTrace();
            }
            System.out.println(11111);
        });

        TimeUnit.SECONDS.sleep(2);
        System.out.println("main end");
    }

Submit tasks

Submit the task if it is not your own custom exception , Need to catch exception , Do exception handling

     try {
    
 pool.execute(() -> {
    

         // Mission ...
       });
     }catch (RejectedExecutionException handler){
    
       // exception handling   According to business   Handle 
       loggerHelper.infoLog(" xxxxxx : " + object + "  The mission was rejected ...");
     }

About CompletableFuture.runAsync Asynchronous processing tasks

CompletableFuture.runAsync The default is to use ForkJoinPool.commonPool() Thread pool , The default is cpu The core number -1 Thread pool thread size (1 individual cpu yes 1 individual cpu)

   if (parallelism < 0 && // default 1 less than #cores
            (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
            parallelism = 1;
        if (parallelism > MAX_CAP)
            parallelism = MAX_CAP;

Default thread pool ( Greater than 1 The core , Maximum number of cores - 1) And the thread exception cannot be caught

CompletableFuture.runAsync exception handling

 CompletableFuture.runAsync(() -> {
    
    System.out.println(" ------- executor --- ");
    System.out.println(" ---------- " + 1 / 0);
}, ExecutorUtil.executor).whenComplete((re,t)->{
    

     System.out.println(" result : " + re);
     System.out.println(t.getMessage());
     t.printStackTrace();
 });

Or in the mission try catch Capture processing