异步线程池

爱被打了一巴掌 2022-08-20 13:10 139阅读 0赞

**异步线程池**

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Collection;
    import java.util.List;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Future;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.RejectedExecutionHandler;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class AsyncThreadUtil {
    
        private final static Logger logger = LoggerFactory.getLogger(AsyncThreadUtil.class);
    
        private static final int corePoolSize = Runtime.getRuntime().availableProcessors() * 50;   // 按进程数 * 50
    
        private static final int maxPoolSize = corePoolSize * 2; 
    
        private static final int maxQueueSize = maxPoolSize * 10; 
        /**
         * 平均每秒任务数量 * 平均执行等待秒数 / 平均可接受任务处理秒数
         * <p>
         * 假设每秒1000个任务
         */
        private static final ExecutorService executorService = new ThreadPoolExecutor(
                corePoolSize, maxPoolSize, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(maxQueueSize),
                new CustomRejectedPolicy());
    
        public static void execute(Runnable command) {
            executorService.execute(command);
        }
    
        public static <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
            try {
                return executorService.invokeAll(tasks);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    
        public static <T> Future<T> submit(Callable<T> task) {
            return executorService.submit(task);
        }
    
        public static void shutdown() {
            try {
                executorService.shutdown();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    
        public static int getQueueSize() {
            return ((ThreadPoolExecutor) executorService).getQueue().size();
        }
    
        private static class CustomRejectedPolicy implements RejectedExecutionHandler {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                logger.info("executorService reject execute: {}", r.getClass().getName());
                if (!executor.isShutdown()) {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                    	logger.error("thread errpr.",e);
                    }
                    executor.execute(r);
                }
            }
        }
    }

测试：

public static void main(String[] args) {
    		
    		AsyncThreadUtil.execute(new Runnable() {
    			@Override
    			public void run() {
    				// TODO do something
    			}
    		});
    	}