public class TaskDisposeUtils {
      /**
     * 使用线程池批处理文件，当所有任务处理完毕后才会返回
     *
     * @param taskList 任务列表
     * @param consumer 处理任务的方法
     * @param executor 线程池
     * @param <T>
     * @throws InterruptedException
     */
    public static <T> void dispose(List<T> taskList, Consumer<? super T> consumer, Executor executor) throws InterruptedException {
        if (taskList == null || taskList.isEmpty() || consumer == null) return;
        CountDownLatch latch = new CountDownLatch(taskList.size());
        for (T item : taskList) {
            executor.execute(() -> {
                try {
                    consumer.accept(item);
                } finally {
                    latch.countDown();
                }
            });
        }
        latch.await();
    }

    public static void main(String[] args) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        List<String> taskList = List.of("短信-1", "短信-2", "短信-3");
        ExecutorService executor = Executors.newFixedThreadPool(10);
        dispose(taskList, TaskDisposeUtils::doTask, executor);
       System.out.println("任务处理完毕,耗时(ms):" + (System.currentTimeMillis() - startTime));
        executor.shutdown();
    }

    public static void doTask(String task) {
        System.out.println(task + "发送成功");
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class LoadRunnerUtils {
    public static class LoadRunnerResult {
        private int requests;           // 请求总数
        private int concurrency;        // 并发量
        private int successRequests;    // 成功请求数
        private int failRequests;       // 失败请求数
        private int timeTakenForTests;  // 请求总耗时(ms)
        private float requestsPerSecond; // 每秒请求数（吞吐量）
        private float timePerRequest;    // 每个请求平均耗时(ms)
        private float fastestCostTime;   // 最快的请求耗时(ms)
        private float slowestCostTime;   // 最慢的请求耗时(ms)
    }

    /**
     * 执行并发压力测试
     * @param requests    总请求数
     * @param concurrency 并发数量
     * @param command     被测试的代码
     * @return 压测结果
     */
    public static LoadRunnerResult run(int requests, int concurrency, Runnable command) throws InterruptedException {
        log.info("压测开始......");
        // 创建固定大小的线程池，预热所有核心线程
        ThreadPoolExecutor executor = createThreadPool(concurrency);
        // 用于等待所有请求完成的同步器
        CountDownLatch latch = new CountDownLatch(requests);  
        // 统计数据（使用原子类保证线程安全）
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger fastestTime = new AtomicInteger(Integer.MAX_VALUE);
        AtomicInteger slowestTime = new AtomicInteger(Integer.MIN_VALUE);
        long startTime = System.currentTimeMillis();
        // 提交所有压测任务
        for (int i = 0; i < requests; i++) {
            executor.execute(() -> executeRequest(command, successCount, fastestTime, slowestTime, latch));
        }
        // 等待所有请求完成
        latch.await();
        executor.shutdown();
        long endTime = System.currentTimeMillis();
        log.info("压测结束，总耗时(ms):{}", (endTime - startTime));
        // 返回压测结果
    }
    
    /**
     * 创建线程池并预热核心线程
     */
    private static ThreadPoolExecutor createThreadPool(int concurrency) {
        // 参数依次为 coolPoolSize、maxPoolSize、keepAliveTime、TimeUnit、等待队列
        ThreadPoolExecutor executor = new ThreadPoolExecutor(concurrency, concurrency,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        executor.prestartAllCoreThreads();
        return executor;
    }
    
    /**
     * 执行单个请求并统计结果
     */
    private static void executeRequest(Runnable command, AtomicInteger successCount, 
                                     AtomicInteger fastestTime, AtomicInteger slowestTime, 
                                     CountDownLatch latch) {
        try {
            long requestStart = System.currentTimeMillis();
            command.run(); // 执行被测试的方法
            int costTime = (int)(System.currentTimeMillis() - requestStart);
            // 更新统计数据
            updateFastestTime(fastestTime, costTime);
            updateSlowestTime(slowestTime, costTime);
            successCount.incrementAndGet();
        } catch (Exception e) {
            log.error("请求执行失败: {}", e.getMessage());
        } finally {
            latch.countDown(); // 标记当前请求完成
        }
    }
    
    /**
     * 原子更新最快耗时
     */
    private static void updateFastestTime(AtomicInteger fastestTime, int costTime) {
        while (true) {
            int current = fastestTime.get();
            if (current <= costTime || fastestTime.compareAndSet(current, costTime)) {
                break;
            }
        }
    }
}

public class TestController {
    /**
     * Juc中的Semaphore可以实现限流功能，可以将 Semaphore 想象成停车场入口的大爷，
     * 大爷手里面拥有一定数量的停车卡（也可以说是令牌），卡的数量是多少呢？就是Semaphore构造方法中指定的，如下就是50个卡，
     * 车主想进去停车，先要从大爷手中拿到一张卡，出来的时候，需要还给大爷，如果拿不到卡，就不能进去停车。
     * semaphore 内部提供了获取令牌，和还令牌的一些方法
     */
    private Semaphore semaphore = new Semaphore(50);
    /**
     * 来个案例，下面是一个下单的方法，这个方法最多只允许 50 个并发，若超过50个并发，则进来的请求，最多等待1秒，如果无法获取到令牌，则快速返回失败，请重试
     * @return
     */
    @GetMapping("/placeOrder")
    public String placeOrder() throws InterruptedException {
        /**
         * semaphore 在上面定义的，里面有50个令牌，也就是同时可以支持50个并发请求
         * 下面的代码，尝试最多等待1秒去获取令牌，获取成功，则进入下单逻辑，获取失败，则返回系统繁忙，请稍后重试
         */
        boolean flag = this.semaphore.tryAcquire(1, 1L, TimeUnit.SECONDS);
        // 获取到令牌，则进入下单逻辑
        if (flag) {
            try {
                //这里休眠2秒，模拟下单的操作
                TimeUnit.SECONDS.sleep(2);
                return "下单成功";
            } finally {
                //这里一定不要漏掉了，令牌用完了，要还回去
                this.semaphore.release();
            }
        } else {
            return "系统繁忙，请稍后重试";
        }
    }
}

public class GoodsController {
    /**
     * 根据商品id获取商品信息(基本信息、描述信息、评论量，收藏量)
     *
     * @param goodsId 商品id
     * @return
     * @throws InterruptedException
     */
    @GetMapping("/getGoodsDetail")
    public GoodsDetailResponse getGoodsDetail(@RequestParam("goodsId") String goodsId) {
        long st = System.currentTimeMillis();
        GoodsDetailResponse goodsDetailResponse = new GoodsDetailResponse();
        // 1、获取商品基本信息，耗时100ms
        goodsDetailResponse.setGoodsInfo(this.getGoodsInfo(goodsId));
        //2、获取商品描述信息，耗时100ms
        goodsDetailResponse.setGoodsDescription(this.getGoodsDescription(goodsId));
        //3、获取商品评论量，耗时100ms
        goodsDetailResponse.setCommentCount(this.getGoodsCommentCount(goodsId));
        //4、获取商品收藏量，耗时100ms
        goodsDetailResponse.setFavoriteCount(this.getGoodsFavoriteCount(goodsId));
        // 总耗时为500ms左右
        LOGGER.info("获取商品信息，普通版耗时：{} ms", (System.currentTimeMillis() - st));
    }
    // 使用sleep模拟数据库查询的延迟
    private int getGoodsFavoriteCount(String goodsId) { 
      try {
          log.info("获取商品收藏量");
          TimeUnit.MILLISECONDS.sleep(100);
      } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw new RuntimeException(e);
      }
      return 10000;
  }

public class GoodsController {
		/**
     * 优化后的方法，根据商品id获取商品信息(基本信息、描述信息、评论量，收藏量)
     *
     * @param goodsId 商品id
     * @return
     * @throws InterruptedException
     */
    @GetMapping("/getGoodsDetailNew")
    public GoodsDetailResponse getGoodsDetailNew(@RequestParam("goodsId") String goodsId) {
        long st = System.currentTimeMillis();
        GoodsDetailResponse goodsDetailResponse = new GoodsDetailResponse();
        // 1、获取商品基本信息，耗时100ms
        CompletableFuture<Void> goodsInfoCf = CompletableFuture.runAsync(() -> goodsDetailResponse.setGoodsInfo(this.getGoodsInfo(goodsId)), this.goodsThreadPool);
        //2、获取商品描述信息，耗时100ms
        CompletableFuture<Void> goodsDescriptionCf = CompletableFuture.runAsync(() -> goodsDetailResponse.setGoodsDescription(this.getGoodsDescription(goodsId)), this.goodsThreadPool);
        //3、获取商品评论量，耗时100ms
        CompletableFuture<Void> goodsCommentCountCf = CompletableFuture.runAsync(() -> goodsDetailResponse.setCommentCount(this.getGoodsCommentCount(goodsId)), this.goodsThreadPool);
        //4、获取商品收藏量，耗时100ms
        CompletableFuture<Void> goodsFavoriteCountCf = CompletableFuture.runAsync(() -> goodsDetailResponse.setFavoriteCount(this.getGoodsFavoriteCount(goodsId)), this.goodsThreadPool);
        //等待上面执行结束
        CompletableFuture.allOf(goodsInfoCf, goodsDescriptionCf, goodsCommentCountCf, goodsFavoriteCountCf).join();
        // 总耗时大约在100ms左右
        LOGGER.info("获取商品信息，使用线程池并行查询耗时：{} ms", (System.currentTimeMillis() - st));
        return goodsDetailResponse;
    }
}

@Bean
public ThreadPoolTaskExecutor goodsThreadPool() {
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setThreadNamePrefix("ThreadPool-Goods-");
    // 核心线程数为cpu核数 * 4，最大线程数据为cpu核数 * 8
    threadPoolTaskExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors() * 4);
    threadPoolTaskExecutor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 8);
    // 队列容量为0，则任务就不会进入队列
    threadPoolTaskExecutor.setQueueCapacity(0);
    // 拒绝策略使用CallerRunsPolicy，让当前线程去兜底去执行任务
    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    return threadPoolTaskExecutor;
}

1、Spring去数据库连接池拿到一个数据库连接
2、开启事务
3、执行bigTransaction()中的代码
4、提交事务
5、将数据库连接还给数据库连接池中

@Transactional
public void bigTransaction() throws InterruptedException {
    // 1、getData()方法模拟一个比较耗时的获取数据的操作，这个方法内部会休眠5秒
    String data = this.getData();
    //2、将上面获取到的数据写入到db中
    Lesson007PO po = new Lesson007PO();
    po.setId(UUID.randomUUID().toString());
    po.setData(data);
    this.lesson007Mapper.insert(po);
}

public String getData() throws InterruptedException {
    //休眠5秒
    TimeUnit.SECONDS.sleep(5);
    return UUID.randomUUID().toString();
}

/**
 * 使用 TransactionTemplate 编程式事务，可以灵活的控制事务的范围
 *
 * @throws InterruptedException
 */
public void smallTransaction() throws InterruptedException {
    // 1、调用getData()方法，讲获取的数据写到db中，假设 getData方法比较耗时，比如耗时 5秒
    String data = this.getData();

    //2、将上面获取到的数据写入到db中
    Lesson007PO po = new Lesson007PO();
    po.setId(UUID.randomUUID().toString());
    po.setData(data);

    // this.transactionTemplate.executeWithoutResult可以传入一个Consumer，这个Consumer表述需要在事务中执行的业务操作
    this.transactionTemplate.executeWithoutResult(action -> {
        this.lesson007Mapper.insert(po);
    });
}

public class ThreadPoolManager { 
     /** 线程池Map */
     private static Map<String, ThreadPoolTaskExecutor> threadPoolMap = new ConcurrentHashMap<String, ThreadPoolTaskExecutor>(16);
		/**
     * 创建新的线程池，如果线程池已经创建，返回已经创建的线程池
     *
     * @param name                     线程池名称
     * @param corePoolSize             核心线程数
     * @param maxPoolSize              最大线程数
     * @param queueCapacity            队列大小
     * @param keepAliveSeconds         线程池存活时间（秒）
     * @param threadFactory            线程工厂
     * @param rejectedExecutionHandler 拒绝策略
     * @return
     */
    public static ThreadPoolTaskExecutor newThreadPool(String name, int corePoolSize, int maxPoolSize, int queueCapacity, int keepAliveSeconds, ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
        return threadPoolMap.computeIfAbsent(name, threadGroupName -> {
            ThreadPoolTaskExecutor threadPoolExecutor = new ThreadPoolTaskExecutor() {
                private boolean initialized = false; // 初始化标记
                @Override
                protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
                    if (queueCapacity > 0) {
                      	// 使用自定义的 ResizeLinkedBlockingQueue 替代默认队列，保证能够在运行时动态调整队列大小
                        return new ResizeLinkedBlockingQueue<>(queueCapacity);
                    } else {
                      	// 当队列容量设置为0时，使用 SynchronousQueue。表示任务必须立即被线程处理，否则就会被拒绝
                        return new SynchronousQueue<>();
                    }
                }
                @Override
                public void setQueueCapacity(int queueCapacity) {
                  	// 确保线程池已初始化且队列类型正确时才调整容量
                    if (this.initialized && this.getThreadPoolExecutor() != null &&
                            this.getThreadPoolExecutor().getQueue() != null &&
                            this.getThreadPoolExecutor().getQueue() instanceof ResizeLinkedBlockingQueue) {
                        ((ResizeLinkedBlockingQueue) this.getThreadPoolExecutor().getQueue()).setCapacity(queueCapacity);
                    }
                    super.setQueueCapacity(queueCapacity);
                }

                @Override
                public void afterPropertiesSet() {
                  	// 使用 initialized 标记确保线程池只被初始化一次
                    if (initialized) {
                        return;
                    }
                    super.afterPropertiesSet();
                    this.initialized = true;
                }
            };
            // 设置参数
            threadPoolExecutor.setCorePoolSize(corePoolSize);
            ...
            if (threadFactory != null) {
                // ThreadFactory相当于线程池的一些规定 (招聘标准)
                threadPoolExecutor.setThreadFactory(threadFactory);
            }
            if (rejectedExecutionHandler != null) {
                // 拒绝执行的策略，上方提到的四种拒绝策略
                threadPoolExecutor.setRejectedExecutionHandler(rejectedExecutionHandler);
            }
            threadPoolExecutor.afterPropertiesSet();
            return threadPoolExecutor;
        });
    }
   
    /** 获取所有线程池信息 */
    public static List<ThreadPoolInfo> threadPoolInfoList() {
        return threadPoolMap
                .entrySet()
                .stream()
                .map(entry -> threadPoolInfo(entry.getKey(), entry.getValue()))
                .collect(Collectors.toList());
    }
 }

/**
 * 设置容量
 * @param capacity
 */
public void setCapacity(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        if (count.get() > capacity) {
            throw new IllegalArgumentException();
        }
        this.capacity = capacity;
    } finally {
        putLock.unlock();
    }
}

public class ThreadPoolManager {  
		/**
     * 动态变更线程池（如：扩缩容、扩缩队列大小）
     * @param threadPoolChange 变更线程池信息
     */
    public static void changeThreadPool(ThreadPoolChange threadPoolChange) {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = threadPoolMap.get(threadPoolChange.getName());
        if (threadPoolTaskExecutor == null) {
            throw new IllegalArgumentException();
        }
        if (threadPoolChange.getCorePoolSize() > threadPoolChange.getMaxPoolSize()) {
            throw new IllegalArgumentException();
        }
        synchronized (ThreadPoolManager.class) {
            // 增加线程情况：需要修改的最大线程大于当前核心线程，则设置当前核心线程为修改后的最大线程
            if (threadPoolChange.getMaxPoolSize() > threadPoolTaskExecutor.getCorePoolSize()) {
              	// 先调整最大线程数，再调整核心数 （先扩大总量，再扩大局部）
                threadPoolTaskExecutor.setMaxPoolSize(threadPoolChange.getMaxPoolSize());
                threadPoolTaskExecutor.setCorePoolSize(threadPoolChange.getCorePoolSize());
                threadPoolTaskExecutor.setQueueCapacity(threadPoolChange.getQueueCapacity());
            } else {
              	// 先减少局部，再减少总量
                threadPoolTaskExecutor.setCorePoolSize(threadPoolChange.getCorePoolSize());
                threadPoolTaskExecutor.setMaxPoolSize(threadPoolChange.getMaxPoolSize());
                threadPoolTaskExecutor.setQueueCapacity(threadPoolChange.getQueueCapacity());
            }
        }
    } 
}

@Component
public class SpringJobRunManager implements CommandLineRunner {
  	// applicationContext 主要用于在任务执行时动态获取和调用指定的 Bean 和方法，实现灵活的任务调度。
    // 可以把它理解为“Spring 管理的对象工厂和服务总线”
    @Autowired
    private ApplicationContext applicationContext;
		// threadPoolTaskScheduler 是 Spring 提供的线程池定时任务调度器，主要用于在应用中以多线程方式执行定时任务（如定时执行、Cron表达式调度等）
    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;
    // job表相关服务
    @Autowired
    private JobService jobService;
    /**
     * 系统重正在运行中的job列表
     */
    private Map<String, SpringJobTask> runningJobMap = new ConcurrentHashMap<>();
    /**
     * springboot应用启动后会回调
     *
     * @param args incoming main method arguments
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        //1、启动job
        this.startAllJob();
        //2、监控db中job的变化（job增、删、改），然后同步给job执行器去执行
        this.monitorDbJobChange();
    }
}

private void startAllJob() {
    List<Job> jobList = this.jobService.getStartJobList();
    for (Job job : jobList) {
        this.startJob(job);
    }
}
/**
 * 启动一个定时任务（job）
 *
 * @param job 需要启动的任务对象
 */
private void startJob(Job job) {
    // 1. 创建任务执行体，注入Spring上下文，便于任务内获取Bean对象
    SpringJobTask springJobTask = new SpringJobTask(job, this.applicationContext);
    // 2. 根据job的cron表达式创建调度触发器
    CronTrigger trigger = new CronTrigger(job.getCron());
    // 3. 使用线程池调度器注册任务，返回调度句柄
    ScheduledFuture<?> scheduledFuture = this.threadPoolTaskScheduler.schedule(springJobTask, trigger);
    // 4. 记录调度句柄到任务对象，便于后续取消 -> 
    springJobTask.setScheduledFuture(scheduledFuture);
    // 5. 将任务放入运行中的任务Map，方便管理和查找
    runningJobMap.put(job.getId(), springJobTask);
    // 6. 记录日志，方便追踪
    logger.info("启动 job 成功:{}", JSONUtil.toJsonStr(job));
}
/**
 * 删除（停止）一个定时任务
 * @param job 需要删除的任务对象
 */
private void deleteJob(Job job) {
    if (job == null) {
        return;
    }
    // 1. 从运行中的任务Map获取任务
    SpringJobTask springJobTask = this.runningJobMap.get(job.getId());
    if (springJobTask == null) { return;}
    // 2. 取消任务调度（停止定时执行）
    springJobTask.getScheduledFuture().cancel(false);
    // 3. 从Map中移除该任务
    runningJobMap.remove(job.getId());
    // 4. 记录日志
    logger.info("移除 job 成功:{}", JSONUtil.toJsonStr(job));
}
/**
 * 更新一个定时任务（先删除再启动）
 * @param job 需要更新的任务对象
 */
public void updateJob(Job job) {
    // 1. 先删除旧的任务
    this.deleteJob(job);
    // 2. 再启动新的任务
    this.startJob(job);
    // 3. 记录日志
    logger.info("更新 job 成功:{}", JSONUtil.toJsonStr(job));
}

/**
 * 监控db中job的变化，每5秒监控一次，这个频率大家使用的时候可以稍微调大点
 */
private void monitorDbJobChange() {
    this.threadPoolTaskScheduler.scheduleWithFixedDelay(this::jobChangeDispose, Duration.ofSeconds(5));
}
// 任务变化处理
private void jobChangeDispose() {
    //1、从db中拿到所有job，和目前内存中正在运行的所有job对比，可得到本次新增的job、删除的job、更新的job
    JobChange jobChange = this.getJobChange();
    //2、处理新增的job
    for (Job job : jobChange.getAddJobList()) { this.startJob(job);}
    //3、处理删除的job
    for (Job job : jobChange.getDeleteJobList()) { this.deleteJob(job);}
    //4、处理变化的job
    for (Job job : jobChange.getUpdateJobList()) { this.updateJob(job);}
}
private JobChange getJobChange() {
    //新增的job
    List<Job> addJobList = new ArrayList<>();
    //删除的job
    List<Job> deleteJobList = new ArrayList<>();
    //更新的job
    List<Job> updateJobList = new ArrayList<>();
    //从db中拿到所有job，和目前内存中正在运行的所有job对比，可得到本次新增的job、删除的job、更新的job
    List<Job> startJobList = this.jobService.getStartJobList();
    for (Job job : startJobList) {
        SpringJobTask springJobTask = runningJobMap.get(job.getId());
        if (springJobTask == null) {
            addJobList.add(job);
        } else {
            //job的执行规则变了
            if (jobIsChange(job, springJobTask.getJob())) {
                updateJobList.add(job);
            }
        }
    }
    //获取被删除的job，springJobTaskMap中存在的，而startJobList不存在的，则是需要从当前运行列表中停止移除的
    Set<String> startJobIdList = CollUtils.convertSet(startJobList, Job::getId);
    for (Map.Entry<String, SpringJobTask> springJobTaskEntry : runningJobMap.entrySet()) {
        if (!startJobIdList.contains(springJobTaskEntry.getKey())) {
            deleteJobList.add(springJobTaskEntry.getValue().getJob());
        }
    }
    //返回job变更结果
    return new JobChange(addJobList, updateJobList, deleteJobList);
}
/**
 * 检测两个job是否发生了变化，（cron、beanName、beanMethod）中有任意一项变动了，则返回true
 *
 * @param job1
 * @param job2
 * @return
 */
private boolean jobIsChange(Job job1, Job job2) {
    return !(Objects.equals(job1.getCron(), job2.getCron()) &&
            Objects.equals(job1.getBeanName(), job2.getBeanName()) &&
            Objects.equals(job1.getBeanMethod(), job2.getBeanMethod()));
}