package com.sjyt.icom.asst.conf;

import com.sjyt.icom.asst.conf.properties.AbstractExecutorPool;
import com.sjyt.icom.asst.conf.properties.LocalExecutorPool;
import com.sjyt.icom.asst.conf.properties.PublisherExecutorPool;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 重新默认线程池配置
 *
 * @author renyanru
 * @version V1.0.0
 * @date 2022/6/10 9:33
 */
@Slf4j
@Aspect
@Configuration
@EnableAsync(proxyTargetClass = true)
public class ThreadPoolConfig implements AsyncConfigurer {
    private final LocalExecutorPool localConfig;
    private final PublisherExecutorPool publisherConfig;

    @Autowired
    public ThreadPoolConfig(LocalExecutorPool localConfig, PublisherExecutorPool publisherConfig) {
        this.localConfig = localConfig;
        this.publisherConfig = publisherConfig;
    }

    /**
     * 重写默认线程池配置
     *
     * @return result
     */
    @Override
    public Executor getAsyncExecutor() {
        return initExecutor(localConfig, localConfig.getThreadNamePrefix(), callerRunsPolicy());
    }

    /**
     * 监听通知线程池
     *
     * @return publisherAsync
     */
    @Bean("publisherAsync")
    public Executor publisherAsync() {
        return initExecutor(publisherConfig, publisherConfig.getThreadNamePrefix(), callerRunsPolicy());
    }

    private Executor initExecutor(AbstractExecutorPool abstractExecutorPool, String threadName, RejectedExecutionHandler rejectedExecutionHandler) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(abstractExecutorPool.getCorePoolSize());
        executor.setMaxPoolSize(abstractExecutorPool.getMaxPoolSize());
        executor.setKeepAliveSeconds(abstractExecutorPool.getKeepAliveSeconds());
        executor.setQueueCapacity(abstractExecutorPool.getQueueCapacity());
        executor.setThreadNamePrefix(threadName);
        executor.setRejectedExecutionHandler(rejectedExecutionHandler);
        executor.initialize();
        return executor;
    }


    /**
     * 当poolSize已达到maxPoolSize，如何处理（是拒绝或者交由其他线程处理）
     * CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的新线程来执行（交由调用方线程执行）
     *
     * @return callerRunsPolicy
     */
    private RejectedExecutionHandler callerRunsPolicy() {
        return new ThreadPoolExecutor.CallerRunsPolicy();
    }

    /**
     * 当poolSize已达到maxPoolSize，如何处理（是拒绝或者交由其他线程处理）
     * AbortPolicy：直接抛出异常
     *
     * @return callerRunsPolicy
     */
    private RejectedExecutionHandler abortPolicy() {
        return new ThreadPoolExecutor.AbortPolicy();
    }

    /**
     * 当poolSize已达到maxPoolSize，如何处理（是拒绝或者交由其他线程处理）
     * DiscardPolicy：直接丢弃
     *
     * @return callerRunsPolicy
     */
    private RejectedExecutionHandler discardPolicy() {

        return new ThreadPoolExecutor.DiscardPolicy();
    }

    /**
     * 当poolSize已达到maxPoolSize，如何处理（是拒绝或者交由其他线程处理）
     * DiscardOldestPolicy：丢弃队列中最老的任务
     *
     * @return callerRunsPolicy
     */
    private RejectedExecutionHandler discardOldestPolicy() {
        return new ThreadPoolExecutor.DiscardOldestPolicy();
    }

    /**
     * 异步任务异常处理
     *
     * @return result
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) -> {
            log.error("===========" + "异常方法:" + method.getName() + "===========", ex);
            log.error("===========" + ex.getMessage() + "===========", ex);
        };
    }
}
