/**
 * 自定義線程池
 * <p>
 * 優(yōu)點(diǎn)：可以自定義參數(shù)
 * </p>
 */
@Test
public void newThreadPoolExecutor() {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
            // 核心線程數(shù)
            3,
            // 最大線程數(shù)
            5,
            // 空閑線程最大存活時(shí)間
            60L,
            // 空閑線程最大存活時(shí)間單位
            TimeUnit.SECONDS,
            // 等待隊(duì)列及大小
            new ArrayBlockingQueue<>(100),
            // 創(chuàng)建新線程時(shí)使用的工廠
            Executors.defaultThreadFactory(),
            // 當(dāng)線程池達(dá)到最大時(shí)的處理策略
//                new ThreadPoolExecutor.AbortPolicy()          // 拋出RejectedExecutionHandler異常
            new ThreadPoolExecutor.CallerRunsPolicy()       // 交由調(diào)用者的線程執(zhí)行
//                new ThreadPoolExecutor.DiscardOldestPolicy()  // 丟掉最早未處理的任務(wù)
//                new ThreadPoolExecutor.DiscardPolicy()        // 丟掉新提交的任務(wù)
    );

    // 總共5個(gè)任務(wù)
    for (int i = 1; i <= 5; i++) {
        int taskIndex = i;
        executor.execute(() -> {
            log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行任務(wù) " + taskIndex);

            // 每個(gè)任務(wù)耗時(shí)1秒
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    executor.shutdown();
}

/**
 * 固定大小線程池
 * <p>
 * 優(yōu)點(diǎn)：當(dāng)任務(wù)執(zhí)行較快，且任務(wù)較少時(shí)使用方便
 * </p>
 * <p>
 * 風(fēng)險(xiǎn)：當(dāng)處理較慢時(shí)，等待隊(duì)列的任務(wù)堆積會(huì)導(dǎo)致OOM
 * </p>
 */
@Test
public void newFixThreadPool() {
    // 3個(gè)固定線程
    ExecutorService executorService = Executors.newFixedThreadPool(3);

    // 總共5個(gè)任務(wù)
    for (int i = 1; i <= 5; i++) {
        int taskIndex = i;
        executorService.execute(() -> {
            log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行任務(wù) " + taskIndex);

            // 每個(gè)任務(wù)耗時(shí)1秒
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    executorService.shutdown();
}

/**
 * Creates a thread pool that reuses a fixed number of threads
 * operating off a shared unbounded queue.  At any point, at most
 * {@code nThreads} threads will be active processing tasks.
 * If additional tasks are submitted when all threads are active,
 * they will wait in the queue until a thread is available.
 * If any thread terminates due to a failure during execution
 * prior to shutdown, a new one will take its place if needed to
 * execute subsequent tasks.  The threads in the pool will exist
 * until it is explicitly {@link ExecutorService#shutdown shutdown}.
 *
 * @param nThreads the number of threads in the pool
 * @return the newly created thread pool
 * @throws IllegalArgumentException if {@code nThreads <= 0}
 */
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

/**
 * 單一線程池
 * <p>
 * 優(yōu)勢(shì)：保存任務(wù)按照提交的順序執(zhí)行
 * </p>
 * <p>
 * 風(fēng)險(xiǎn)：當(dāng)處理較慢時(shí)，等待隊(duì)列的任務(wù)堆積會(huì)導(dǎo)致OOM
 * </p>
 */
@Test
public void newSingleThreadExecutor() {
    // 1個(gè)線程
    ExecutorService executor = Executors.newSingleThreadExecutor();

    // 總共5個(gè)任務(wù)
    for (int i = 1; i <= 5; i++) {
        int taskIndex = i;
        executor.execute(() -> {
            log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行任務(wù) " + taskIndex);

            // 每個(gè)任務(wù)耗時(shí)1秒
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    executor.shutdown();
}

/**
 * Creates an Executor that uses a single worker thread operating
 * off an unbounded queue. (Note however that if this single
 * thread terminates due to a failure during execution prior to
 * shutdown, a new one will take its place if needed to execute
 * subsequent tasks.)  Tasks are guaranteed to execute
 * sequentially, and no more than one task will be active at any
 * given time. Unlike the otherwise equivalent
 * {@code newFixedThreadPool(1)} the returned executor is
 * guaranteed not to be reconfigurable to use additional threads.
 *
 * @return the newly created single-threaded Executor
 */
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

/**
 * 共享線程池
 * <p>
 * 優(yōu)勢(shì)：當(dāng)在某一時(shí)間段內(nèi)任務(wù)較多，且執(zhí)行較快時(shí)方便使用
 * </p>
 * <p>
 * 風(fēng)險(xiǎn)：當(dāng)處理較慢時(shí)，會(huì)創(chuàng)建大量的線程
 * </p>
 */
@Test
public void newCachedThreadPool() {
    ExecutorService executor = Executors.newCachedThreadPool();

    // 總共5個(gè)任務(wù)
    for (int i = 1; i <= 5; i++) {
        int taskIndex = i;
        executor.execute(() -> {
            log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行任務(wù) " + taskIndex);

            // 每個(gè)任務(wù)耗時(shí)1秒
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    executor.shutdown();
}

/**
 * Creates a thread pool that creates new threads as needed, but
 * will reuse previously constructed threads when they are
 * available.  These pools will typically improve the performance
 * of programs that execute many short-lived asynchronous tasks.
 * Calls to {@code execute} will reuse previously constructed
 * threads if available. If no existing thread is available, a new
 * thread will be created and added to the pool. Threads that have
 * not been used for sixty seconds are terminated and removed from
 * the cache. Thus, a pool that remains idle for long enough will
 * not consume any resources. Note that pools with similar
 * properties but different details (for example, timeout parameters)
 * may be created using {@link ThreadPoolExecutor} constructors.
 *
 * @return the newly created thread pool
 */
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

/**
 * 定時(shí)線程池
 * <p>
 * 優(yōu)點(diǎn)：可以定時(shí)執(zhí)行某些任務(wù)
 * </p>
 * <p>
 * 風(fēng)險(xiǎn)：當(dāng)處理較慢時(shí)，等待隊(duì)列的任務(wù)堆積會(huì)導(dǎo)致OOM
 * </p>
 */
@Test
public void newScheduledThreadPool() {
//        // 單一線程
//        ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    // 指定核心線程數(shù)
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(3);

    executor.schedule(() -> {
        log.info("3秒后開始執(zhí)行，以后不再執(zhí)行");

        // 每個(gè)任務(wù)耗時(shí)1秒
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, 3, TimeUnit.SECONDS);
//
//        executor.scheduleAtFixedRate(() -> {
//            log.info("3秒后開始執(zhí)行，以后每2秒執(zhí)行一次");
//
//            // 每個(gè)任務(wù)耗時(shí)1秒
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }, 3, 2, TimeUnit.SECONDS);
//
//        executor.scheduleWithFixedDelay(() -> {
//            log.info("3秒后開始執(zhí)行，以后延遲2秒執(zhí)行一次");
//
//            // 每個(gè)任務(wù)耗時(shí)1秒
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }, 3, 2, TimeUnit.SECONDS);
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 自定義線程池配置類
 *
 * @author CL
 */
@Configuration
public class TaskExecutorConfig {

    /**
     * 自定義任務(wù)執(zhí)行器
     *
     * @return {@link TaskExecutor}
     */
    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心線程數(shù)，默認(rèn)1
        int corePoolSize = Runtime.getRuntime().availableProcessors();
        executor.setCorePoolSize(corePoolSize);
        // 最大線程數(shù)，默認(rèn)Integer.MAX_VALUE
        executor.setMaxPoolSize(corePoolSize * 2 + 1);
        // 空閑線程最大存活時(shí)間，默認(rèn)60秒
        executor.setKeepAliveSeconds(3);
        // 等待隊(duì)列及大小，默認(rèn)Integer.MAX_VALUE
        executor.setQueueCapacity(500);
        // 線程的名稱前綴，默認(rèn)該Bean名稱簡寫：org.springframework.util.ClassUtils.getShortName(java.lang.Class<?>)
        executor.setThreadNamePrefix("custom-thread-");
        // 當(dāng)線程池達(dá)到最大時(shí)的處理策略，默認(rèn)拋出RejectedExecutionHandler異常
//        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());         // 拋出RejectedExecutionHandler異常
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());    // 交由調(diào)用者的線程執(zhí)行
//        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); // 丟掉最早未處理的任務(wù)
//        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());       // 丟掉新提交的任務(wù)
        // 等待所有任務(wù)結(jié)束后再關(guān)閉線程池，默認(rèn)false
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 等待所有任務(wù)結(jié)束最長等待時(shí)間，默認(rèn)0毫秒
        executor.setAwaitTerminationSeconds(10);
        // 執(zhí)行初始化
        executor.initialize();
        return executor;
    }

}

/**
 * 示例Service
 *
 * @author CL
 */
public interface DemoService {

    /**
     * 示例方法
     *
     * @return {@link String}
     */
    void demo();

}

import lombok.extern.slf4j.Slf4j;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;


/**
 * 示例Service實(shí)現(xiàn)
 *
 * @author CL
 */
@Slf4j
@Service
public class DemoServiceImpl implements DemoService {

    @Resource
    private TaskExecutor taskExecutor;

    /**
     * 示例方法
     */
    @Override
    public void demo() {
        taskExecutor.execute(() -> {
            log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行Service中的方法");
        });
    }

}

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;

/**
 * 示例異步任務(wù)
 *
 * @author CL
 */
@Slf4j
@Component
@EnableAsync
public class DemoAsync {

    /**
     * 示例方法
     */
    @Async(value = "taskExecutor")
    public void demo() {
        log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行Async中的方法");
    }

}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 自定義定時(shí)任務(wù)調(diào)度配置類
 *
 * @author CL
 */
@Configuration
public class SheduledConfig implements SchedulingConfigurer {

    /**
     * 配置定時(shí)任務(wù)
     *
     * @param scheduledTaskRegistrar 配置任務(wù)注冊(cè)器
     */
    @Override
    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.setScheduler(taskScheduler());

//        // 第二種方式
//        scheduledTaskRegistrar.setScheduler(scheduledExecutorService());
    }

    /**
     * 自定義任務(wù)調(diào)度器
     *
     * @return {@link TaskScheduler}
     */
    @Bean
    public TaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler executor = new ThreadPoolTaskScheduler();
        executor.setPoolSize(5);
        executor.setThreadNamePrefix("custom-scheduler-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

//    /**
//     * 自定義任務(wù)線程池
//     *
//     * @return {@link ScheduledExecutorService}
//     */
//    @Bean
//    public ScheduledExecutorService scheduledExecutorService() {
//        return Executors.newScheduledThreadPool(5);
//    }

}

import com.c3tones.async.DemoAsync;
import com.c3tones.service.DemoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * 示例Controller
 *
 * @author CL
 */
@Slf4j
@RestController
public class DemoController {

    @Resource
    private DemoService demoService;

    @Resource
    private DemoAsync demoAsync;

    /**
     * Service示例方法
     *
     * @return {@link String}
     */
    @RequestMapping("/service")
    public void service() {
        log.info("Service示例方法開始執(zhí)行");
        demoService.demo();
        log.info("Service示例方法結(jié)束執(zhí)行");
    }

    /**
     * 異步示例方法
     *
     * @return {@link String}
     */
    @RequestMapping("/async")
    public void async() {
        log.info("異步示例方法開始執(zhí)行");
        demoAsync.demo();
        log.info("異步示例方法結(jié)束執(zhí)行");
    }

}

curl http://127.0.0.1:8080/service

curl http://127.0.0.1:8080/async

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * 示例定時(shí)任務(wù)
 *
 * @author CL
 */
@Slf4j
@Component
@EnableScheduling
public class DemoScheduled {

    /**
     * 示例方法
     */
    @Scheduled(cron = "0/3 * * * * ? ")
    public void demo() {
        log.info("線程 " + Thread.currentThread().getName() + " 正在執(zhí)行Scheduled中的方法");
    }

}