异步线程池、异步执行器

package org.example.thread;
 
public class Test {
 
    public static void main(String[] args) {
        AsyncProcessExecutor normalExecutor = AsyncProcessExecutorFactory.createNormalExecutor();
        normalExecutor.put(() -> {
            System.out.println("123");
            test();
        });
        normalExecutor.execute();
        System.out.println("00000000000000000000000000000");
    }
 
    private static void test(){
        try {
            Thread.sleep(1000);
            System.out.println("test...............");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}复制

线程池代码

package org.example.thread;
 
/**
 * 异步进程 执行器
 *
 * @date 2021年7月15日13:43:37
 */
public interface AsyncProcessExecutor {
 
 
    /**
     * 存放任务
     *
     * @param task 任务
     * @return AsyncProcessExecutor
     */
    AsyncProcessExecutor put(final Runnable task);
 
    /**
     * 执行
     *
     * @return boolean
     */
    boolean execute();
 
}复制

package org.example.thread;
 
import cn.hutool.core.collection.CollUtil;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
 
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
 
/**
 * 多线程锁执行器 正常处理
 *
 * @date 2020-12-10 10:36
 */
@Slf4j
public class AsyncProcessExecutorByNormal implements AsyncProcessExecutor {
 
    /**
     * 线程池字典
     */
    private static final Map<String, AsyncProcessor> EXECUTOR_MAP = Maps.newConcurrentMap();
 
    /**
     * 线程Key
     */
    private final String key;
 
    /**
     * 任务队列
     */
    private final List<Runnable> taskList;
 
    /**
     * 执行器
     */
    private final AsyncProcessor processor;
 
    /**
     * 构造函数
     */
    public AsyncProcessExecutorByNormal() {
        this.key = "def";
        taskList = new ArrayList<>();
        processor = AsyncProcessExecutorByNormal.getProcessor(this.key);
    }
 
    /**
     * 构造函数
     *
     * @param key 线程池唯一Key
     */
    public AsyncProcessExecutorByNormal(String key) {
        this.key = key;
        taskList = new ArrayList<>();
        processor = AsyncProcessExecutorByNormal.getProcessor(this.key);
    }
 
    /**
     * 获得执行器
     *
     * @param key Key
     * @return AsyncProcessor
     */
    private synchronized static AsyncProcessor getProcessor(String key) {
        AsyncProcessor asyncProcessor = EXECUTOR_MAP.get(key);
        if (null == asyncProcessor) {
            asyncProcessor = new AsyncProcessor();
            asyncProcessor.init(key);
            EXECUTOR_MAP.put(key, asyncProcessor);
        }
        return asyncProcessor;
    }
 
    /**
     * 执行
     *
     * @param task 任务
     */
    @Override
    public AsyncProcessExecutorByNormal put(final Runnable task) {
        taskList.add(task);
        return this;
    }
 
    // ====================================
 
    /**
     * 执行 线程锁 等待查询结果 结果完成后继续执行
     *
     * @return boolean 最终直接结果
     */
    @Override
    public boolean execute() {
        if (CollUtil.isEmpty(this.taskList)) {
            return true;
        }
 
        for (Runnable task : this.taskList) {
            // 多线程执行任务
            this.execute(task);
        }
 
        // 返回执行结果
        return true;
    }
 
    /**
     * 执行指定的任务
     *
     * @param task 任务
     * @return boolean
     */
    private boolean execute(final Runnable task) {
        return processor.executeTask(task);
    }
 
}复制

package org.example.thread;
 
import cn.hutool.core.collection.CollUtil;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
 
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
 
/**
 * 多线程锁执行器
 * 用于当前方法中复杂业务多线程处理，等待线程执行完毕后 获得统一结果
 * 2021年11月2日14:07:54 重构 多线程异步等待执行器
 *
 * @author 周鹏程
 * @date 2020-12-10 10:36
 */
@Slf4j
public class AsyncProcessExecutorByWait implements AsyncProcessExecutor {
 
    /**
     * 线程池字典
     */
    private static final Map<String, AsyncProcessor> EXECUTOR_MAP = Maps.newConcurrentMap();
 
    /**
     * 线程Key
     */
    private final String key;
    /**
     * 任务队列
     */
    private final List<Callable<Object>> taskList;
    /**
     * 执行器
     */
    private final AsyncProcessor processor;
    /**
     * 任务执行计数器
     */
    private AtomicInteger count;
 
 
    /**
     * 构造函数
     */
    public AsyncProcessExecutorByWait() {
        this.key = "def";
        taskList = new ArrayList<>();
        processor = getProcessor(this.key);
    }
 
    /**
     * 构造函数
     *
     * @param key 线程池唯一Key
     */
    public AsyncProcessExecutorByWait(String key) {
        this.key = key;
        taskList = new ArrayList<>();
        processor = getProcessor(this.key);
    }
 
    /**
     * 获得执行器
     *
     * @param key Key
     * @return AsyncProcessor
     */
    private synchronized static AsyncProcessor getProcessor(String key) {
        AsyncProcessor asyncProcessor = EXECUTOR_MAP.get(key);
        if (null == asyncProcessor) {
            asyncProcessor = new AsyncProcessor();
            asyncProcessor.init(key);
            EXECUTOR_MAP.put(key, asyncProcessor);
        }
        return asyncProcessor;
    }
 
    /**
     * 放入执行任务
     * 特殊处理 Runnable 转换为 Callable
     *
     * @param task 任务
     */
    @Override
    public AsyncProcessExecutor put(final Runnable task) {
        taskList.add(Executors.callable(task));
        return this;
    }
 
    /**
     * 执行 线程锁 等待查询结果 结果完成后继续执行
     */
    @Override
    public boolean execute() {
        if (CollUtil.isEmpty(this.taskList)) {
            return true;
        }
 
        // 初始化锁参数
        count = new AtomicInteger(this.taskList.size());
        // 门闩 线程锁
        CountDownLatch latch = new CountDownLatch(this.taskList.size());
 
        for (Callable<Object> task : this.taskList) {
            // 回调减 门闩
            processor.executeTaskAndCallback(task, (result) -> {
                if (result.getSuccess()) {
                    count.decrementAndGet();
                }
                latch.countDown();
                return null;
            });
        }
 
        // 线程锁 等待查询结果 结果完成后继续执行
        try {
            latch.await();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            this.taskList.clear();
        }
 
        // 返回执行结果
        return count.get() == 0;
    }
 
}复制

package org.example.thread;
 
/**
 * 异步进程 执行器 工厂
 *
 * @date 2021年7月15日13:43:37
 */
public final class AsyncProcessExecutorFactory {
 
    private AsyncProcessExecutorFactory() {
    }
 
    /**
     * 创建等待执行器
     *
     * @return AsyncProcessExecutor
     */
    public static AsyncProcessExecutor createWaitExecutor() {
        return new AsyncProcessExecutorByWait();
    }
 
    /**
     * 创建等待执行器
     *
     * @param key KEY
     * @return AsyncProcessExecutor
     */
    public static AsyncProcessExecutor createWaitExecutor(String key) {
        return new AsyncProcessExecutorByWait(key);
    }
 
    /**
     * 创建正常执行器
     *
     * @return AsyncProcessExecutor
     */
    public static AsyncProcessExecutor createNormalExecutor() {
        return new AsyncProcessExecutorByNormal();
    }
 
    // =====================
 
    /**
     * 创建正常执行器
     *
     * @param key KEY
     * @return AsyncProcessExecutor
     */
    public static AsyncProcessExecutor createNormalExecutor(String key) {
        return new AsyncProcessExecutorByNormal(key);
    }
 
}复制

package org.example.thread;
 
import cn.hutool.core.util.StrUtil;
import com.google.common.util.concurrent.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
 
import java.util.concurrent.Callable;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
 
/**
 * 自定义线程执行器 - 等待线程执行完毕不拒绝
 *
 * @author 周鹏程
 * @date 2020-10-08 10:24
 */
@Slf4j
public class AsyncProcessor {
 
    /**
     * 线程池名称格式
     */
    private static final String THREAD_POOL_NAME = "AsyncProcessorWaitPool-{}-%d";
 
    /**
     * 默认线程池关闭等待时间 秒
     */
    private static final int DEFAULT_WAIT_TIME = 10;
 
    /**
     * 线程池监听执行器
     */
    private ListeningExecutorService execute;
 
    /**
     * 初始化
     *
     * @param key 线程池标识
     */
    public void init(String key) {
        if (StringUtils.isBlank(key)) {
            return;
        }
 
        // 线程工厂名称
        String formatThreadPoolName = StrUtil.format(THREAD_POOL_NAME, key);
 
        // 创建 Executor
        // 此处默认最大值改为处理器数量的 4 倍
        try {
            // 监听执行器
            execute = MoreExecutors.listeningDecorator(
                    ThreadPoolFactory.createDefThreadPool(formatThreadPoolName));
 
            // 这里不会自动关闭线程， 当线程超过阈值时 抛异常
            // 关闭事件的挂钩
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                log.info("ProcessorWait 异步处理器关闭");
 
                execute.shutdown();
 
                try {
                    // 等待1秒执行关闭
                    if (!execute.awaitTermination(DEFAULT_WAIT_TIME, TimeUnit.SECONDS)) {
                        log.error("ProcessorWait 由于等待超时，异步处理器立即关闭");
                        execute.shutdownNow();
                    }
                } catch (InterruptedException e) {
                    log.error("ProcessorWait 异步处理器关闭中断");
                    execute.shutdownNow();
                }
 
                log.info("ProcessorWait 异步处理器关闭完成");
            }));
        } catch (Exception e) {
            log.error("ProcessorWait 异步处理器初始化错误", e);
            throw new ExceptionInInitializerError(e);
        }
    }
 
 
    /**
     * 执行任务，不管是否成功<br>
     * 其实也就是包装以后的 {@link } 方法
     *
     * @param task 任务
     * @return boolean
     */
    public boolean executeTask(Runnable task) {
        try {
            execute.execute(task);
        } catch (RejectedExecutionException e) {
            log.error("AsyncProcessorWait 执行任务被拒绝", e);
            return false;
        }
        return true;
    }
 
    /**
     * 提交任务，并可以在稍后获取其执行情况<br>
     * 当提交失败时，会抛出 {@link }
     *
     * @param task 任务
     */
    public <T> void executeTaskAndCallback(Callable<T> task, Function<CallbackResult<T>, Void> callback) {
        ListenableFuture<T> future = execute.submit(task);
        Futures.addCallback(future, new FutureCallback<T>() {
            @Override
            public void onSuccess(T result) {
                CallbackResult<T> callbackResult = new CallbackResult<>();
                callbackResult.setSuccess(true);
                callbackResult.setResult(result);
                // 线程池失败后 返回该 Runnable
                callback.apply(callbackResult);
            }
 
            @Override
            public void onFailure(Throwable t) {
                log.error("线程名称：{} - 执行异常信息：{}", Thread.currentThread().getName(), t.getMessage());
                CallbackResult<T> callbackResult = new CallbackResult<>();
                callbackResult.setSuccess(false);
                callback.apply(callbackResult);
            }
        }, execute);
    }
 
    // =================
 
    /**
     * 回调结果
     *
     * @param <T>
     */
    @Data
    public static class CallbackResult<T> {
 
        /**
         * 状态
         */
        private Boolean success;
 
        /**
         * 结果
         */
        private T result;
 
    }
}复制

package org.example.thread;
 
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
 
import java.util.Map;
import java.util.concurrent.ExecutorService;
 
/**
 * 单线程池
 *
 * @author 周鹏程
 * @date 2021/8/27 17:00
 */
@Slf4j
public final class SyncProcessSingleExecutor {
 
    private static final Map<String, ExecutorService> EXECUTOR_MAP = Maps.newConcurrentMap();
 
    private static final String KEY = "def";
 
    private SyncProcessSingleExecutor() {
    }
 
    /**
     * 执行器
     *
     * @param r 任务
     */
    public static synchronized void execute(Runnable r) {
        execute(KEY, r);
    }
 
    /**
     * 执行器
     *
     * @param key 唯一Key
     * @param r   任务
     */
    public static synchronized void execute(String key, Runnable r) {
        if (null == r) {
            return;
        }
 
        ExecutorService executorService = EXECUTOR_MAP.get(key);
        if (null == executorService) {
            executorService = ThreadUtil.newSingleExecutor();
            EXECUTOR_MAP.put(key, executorService);
        }
 
        executorService.execute(new TaskWrapper(r));
    }
 
    /**
     * Task 包装类<br>
     * 此类型的意义是记录可能会被 Executor 吃掉的异常<br>
     */
    private static class TaskWrapper implements Runnable {
 
        private final Runnable gift;
 
        public TaskWrapper(final Runnable target) {
            this.gift = target;
        }
 
        @Override
        public void run() {
            // 捕获异常，避免在 Executor 里面被吞掉了
            if (gift != null) {
                try {
                    gift.run();
                } catch (Exception e) {
                    String errMsg = StrUtil.format("线程池-包装的目标执行异常: {}", e.getMessage());
                    log.error(errMsg, e);
                }
            }
        }
    }
}复制

package org.example.thread;
 
 
import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
/**
 * 线程池工厂
 *
 * @date 2021/11/2 10:48
 */
public final class ThreadPoolFactory {
 
    /**
     * 默认最大并发数<br>
     */
    private static final int DEFAULT_MAX_CONCURRENT = Runtime.getRuntime().availableProcessors() * 2;
 
    /**
     * 默认线程存活时间
     */
    private static final long DEFAULT_KEEP_ALIVE = 60L;
 
    /**
     * 默认队列大小
     */
    private static final int DEFAULT_SIZE = 1024;
 
    /**
     * 线程池名称格式
     */
    private static final String DEFAULT_THREAD_POOL_NAME = "ProcessPool-{}-%d";
 
 
    private ThreadPoolFactory() {
    }
 
    /**
     * 创建默认的线程池
     *
     * @return ThreadPoolExecutor
     */
    public static ThreadPoolExecutor createDefThreadPool() {
        return createInitThreadPool(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
                TimeUnit.SECONDS, DEFAULT_SIZE, DEFAULT_THREAD_POOL_NAME, new ThreadPoolExecutor.CallerRunsPolicy());
    }
 
    /**
     * 创建默认的线程池
     *
     * @param poolName 线程池名称
     * @return ThreadPoolExecutor
     */
    public static ThreadPoolExecutor createDefThreadPool(String poolName) {
        return createInitThreadPool(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
                TimeUnit.SECONDS, DEFAULT_SIZE, poolName, new ThreadPoolExecutor.CallerRunsPolicy());
    }
 
    /**
     * 创建默认的线程池
     *
     * @param maxConcurrent 最大线程数
     * @param poolName      线程池名称
     * @return ThreadPoolExecutor
     */
    public static ThreadPoolExecutor createDefThreadPool(int maxConcurrent, String poolName) {
        return createInitThreadPool(maxConcurrent, maxConcurrent * 4, DEFAULT_KEEP_ALIVE,
                TimeUnit.SECONDS, DEFAULT_SIZE, poolName, new ThreadPoolExecutor.CallerRunsPolicy());
    }
 
    /**
     * 创建线程池
     *
     * @param coreConcurrent 核心线程数
     * @param maxConcurrent  最大线程数
     * @param keepAlive      线程存活时效
     * @param timeUnit       线程存活单位
     * @param queueSize      队列大小
     * @param poolName       线程池名称
     * @param handler        拒绝处理策略
     * @return ThreadPoolExecutor
     */
    public static ThreadPoolExecutor createInitThreadPool(final int coreConcurrent,
                                                          final int maxConcurrent,
                                                          final long keepAlive,
                                                          final TimeUnit timeUnit,
                                                          final int queueSize,
                                                          final String poolName,
                                                          final RejectedExecutionHandler handler
    ) {
        return new ThreadPoolExecutor(coreConcurrent, maxConcurrent, keepAlive, timeUnit,
                new LinkedBlockingDeque<>(queueSize),
                new ThreadFactoryBuilder().setNameFormat(poolName).build(),
                handler
        );
    }
 
}复制