package com.njzscloud.common.core.thread;


import cn.hutool.core.lang.Assert;
import com.njzscloud.common.core.ex.Exceptions;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.function.Supplier;

@Slf4j
public class ThreadPool {
    private static volatile ThreadPoolExecutor THREAD_POOL;

    /**
     * 默认线程池（虚拟线程）
     */
    public static ThreadPoolExecutor defaultThreadPool() {
        if (THREAD_POOL == null) {
            synchronized (ThreadPool.class) {
                if (THREAD_POOL == null) {
                    THREAD_POOL = createThreadPool("Default");
                }
            }
        }
        return THREAD_POOL;
    }

    /**
     * 创建线程池（平台线程）
     *
     * @param poolName   线程池名称
     * @param poolSize   线程数量
     * @param bufferSize 队列缓冲区大小
     */
    public static ThreadPoolExecutor createThreadPool(String poolName, int poolSize, int bufferSize) {
        return createPlatformThreadPool(poolName, 0, poolSize, 60, bufferSize, Integer.MAX_VALUE, null);
    }

    /**
     * 创建线程池（虚拟线程）
     *
     * @param poolName 线程池名称
     */
    public static ThreadPoolExecutor createThreadPool(String poolName) {
        return createVirtualThreadPool(poolName, null);
    }


    /**
     * 创建线程池（虚拟线程）
     *
     * @param poolName                 线程池名称
     * @param rejectedExecutionHandler 拒绝策略
     */
    public static ThreadPoolExecutor createVirtualThreadPool(String poolName, RejectedExecutionHandler rejectedExecutionHandler) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            new DefaultThreadFactory(true, poolName),
            rejectedExecutionHandler == null ? new ThreadPoolExecutor.AbortPolicy() : rejectedExecutionHandler);
        threadPoolExecutor.allowCoreThreadTimeOut(true);
        return threadPoolExecutor;
    }


    /**
     *
     * @param poolName                 线程池名称
     * @param corePoolSize             核心线程数
     * @param maxPoolSize              最大线程数
     * @param keepAliveSeconds         空闲时间（秒）
     * @param windowCapacity           一级队列大小
     * @param standbyCapacity          二级队列大小
     * @param rejectedExecutionHandler 拒绝策略
     */
    public static ThreadPoolExecutor createPlatformThreadPool(String poolName,
                                                              int corePoolSize, int maxPoolSize, long keepAliveSeconds,
                                                              int windowCapacity, int standbyCapacity,
                                                              RejectedExecutionHandler rejectedExecutionHandler
    ) {
        Assert.isTrue(windowCapacity > 0 && standbyCapacity > 0 && windowCapacity <= standbyCapacity, () -> Exceptions.error("队列大小设置错误"));
        Q<Runnable> q = new Q<>(windowCapacity, standbyCapacity);
        RejectedExecutionHandler abortPolicy = rejectedExecutionHandler == null ? new ThreadPoolExecutor.AbortPolicy() : rejectedExecutionHandler;
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            corePoolSize, maxPoolSize, keepAliveSeconds <= 0 ? 60 : keepAliveSeconds, TimeUnit.SECONDS,
            q,
            new DefaultThreadFactory(false, poolName),
            (r, p) -> {
                if (!q.offerStandby(r)) {
                    log.debug("任务队列已满");
                    abortPolicy.rejectedExecution(r, p);
                } else {
                    log.debug("任务已加入备用队列");
                }
            });
        threadPoolExecutor.allowCoreThreadTimeOut(corePoolSize == 0);
        return threadPoolExecutor;
    }

    public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return CompletableFuture.runAsync(runnable, defaultThreadPool());
    }

    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return CompletableFuture.supplyAsync(supplier, defaultThreadPool());
    }
}
