Spring 定时任务与异步处理

@Configuration
@EnableScheduling  // 启用定时任务支持
public class SchedulingConfig {
}

@Component
public class ScheduledTasks {

    // 固定频率：每隔 5 秒执行（从上次开始时间算起）
    @Scheduled(fixedRate = 5000)
    public void reportCurrentTimeWithFixedRate() {
        System.out.println("fixedRate 执行：" + LocalTime.now());
    }

    // 固定延迟：上次执行完毕后等 5 秒再执行
    @Scheduled(fixedDelay = 5000)
    public void reportCurrentTimeWithFixedDelay() {
        System.out.println("fixedDelay 执行：" + LocalTime.now());
    }

    // Cron 表达式：每天凌晨 2 点执行
    @Scheduled(cron = "0 0 2 * * ?")
    public void dailyCleanup() {
        System.out.println("凌晨清理任务执行");
    }

    // 初始延迟 + 固定频率：启动后 10 秒首次执行，之后每隔 30 秒执行
    @Scheduled(initialDelay = 10000, fixedRate = 30000)
    public void delayedStartTask() {
        System.out.println("延迟启动任务执行：" + LocalTime.now());
    }
}

┌──────── 秒（0-59）
│ ┌────── 分（0-59）
│ │ ┌──── 时（0-23）
│ │ │ ┌── 日（1-31）
│ │ │ │ ┌─ 月（1-12）
│ │ │ │ │ ┌ 星期（0-7，0 和 7 都代表周日）
│ │ │ │ │ │
* * * * * ?

// 问题演示：两个定时任务互相阻塞
@Component
public class BlockingTasks {

    @Scheduled(fixedRate = 2000)
    public void taskA() throws InterruptedException {
        Thread.sleep(5000); // 模拟耗时操作
        System.out.println("任务A 完成 - " + Thread.currentThread().getName());
    }

    @Scheduled(fixedRate = 2000)
    public void taskB() {
        System.out.println("任务B 完成 - " + Thread.currentThread().getName());
    }
    // 结果：任务B 被任务A 阻塞，无法按时执行
}

@Configuration
@EnableScheduling
public class SchedulingConfig implements SchedulingConfigurer {

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        // 设置线程池，核心线程数根据定时任务数量调整
        taskRegistrar.setScheduler(scheduledExecutorService());
    }

    @Bean(destroyMethod = "shutdown")
    public ScheduledExecutorService scheduledExecutorService() {
        return Executors.newScheduledThreadPool(
            4,  // 核心线程数，建议 >= 定时任务数量
            new CustomThreadFactory("scheduled-task-")
        );
    }

    // 自定义线程工厂，便于线程命名和问题排查
    static class CustomThreadFactory implements ThreadFactory {
        private final AtomicInteger counter = new AtomicInteger(1);
        private final String prefix;

        CustomThreadFactory(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, prefix + counter.getAndIncrement());
            thread.setDaemon(true);  // 设为守护线程，不阻止 JVM 关闭
            return thread;
        }
    }
}

@Component
public class DistributedScheduledTask {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Scheduled(cron = "0 0 2 * * ?")
    public void dailyCleanup() {
        // 尝试获取分布式锁，锁 30 分钟自动过期
        String lockKey = "scheduled:daily-cleanup";
        String lockValue = UUID.randomUUID().toString();
        Boolean acquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, Duration.ofMinutes(30));

        if (Boolean.TRUE.equals(acquired)) {
            try {
                // 执行业务逻辑
                doCleanup();
            } finally {
                // 释放锁（使用 Lua 脚本保证原子性，防止误删其他实例的锁）
                releaseLock(lockKey, lockValue);
            }
        } else {
            System.out.println("其他实例已获取锁，跳过本次执行");
        }
    }

    private void releaseLock(String lockKey, String lockValue) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                        "then return redis.call('del', KEYS[1]) " +
                        "else return 0 end";
        redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            List.of(lockKey),
            lockValue
        );
    }
}

@Configuration
@EnableAsync  // 启用异步方法执行支持
public class AsyncConfig {
}

@Service
public class EmailService {

    // 异步发送邮件，调用方无需等待
    @Async
    public void sendEmail(String to, String subject, String content) {
        // 模拟耗时操作
        Thread.sleep(3000);
        System.out.println("邮件已发送至：" + to);
    }
}

@RestController
public class UserController {

    @Autowired
    private EmailService emailService;

    @PostMapping("/register")
    public Result register(@RequestBody UserDTO user) {
        // 1. 同步：保存用户
        userService.save(user);
        // 2. 异步：发送欢迎邮件（不阻塞响应）
        emailService.sendEmail(user.getEmail(), "欢迎注册", "...");
        // 3. 立即返回
        return Result.success("注册成功");
    }
}

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);       // 核心线程数
        executor.setMaxPoolSize(20);       // 最大线程数
        executor.setQueueCapacity(100);    // 队列容量
        executor.setKeepAliveSeconds(60);  // 空闲线程存活时间
        executor.setThreadNamePrefix("async-task-");  // 线程名前缀，便于排查
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  // 拒绝策略
        executor.setWaitForTasksToCompleteOnShutdown(true);  // 关闭时等待任务完成
        executor.setAwaitTerminationSeconds(60);  // 最大等待时间
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        // 异步方法的异常处理器
        return (throwable, method, params) -> {
            System.err.println("异步方法异常 - 方法：" + method.getName()
                + "，参数：" + Arrays.toString(params)
                + "，异常：" + throwable.getMessage());
        };
    }
}

@Service
public class ReportService {

    // 返回 CompletableFuture（推荐）
    @Async
    public CompletableFuture<String> generateReport(String reportId) {
        try {
            // 模拟耗时报表生成
            Thread.sleep(5000);
            String result = "报表-" + reportId + "-生成完成";
            return CompletableFuture.completedFuture(result);
        } catch (InterruptedException e) {
            return CompletableFuture.failedFuture(e);
        }
    }

    // 返回 Future（传统方式）
    @Async
    public Future<String> generateReportLegacy(String reportId) {
        String result = "报表-" + reportId + "-生成完成";
        return new AsyncResult<>(result);
    }
}

@Service
public class OrderService {

    @Autowired
    private ReportService reportService;

    public void processOrder(String orderId) throws Exception {
        // 发起异步调用
        CompletableFuture<String> future = reportService.generateReport(orderId);

        // 方式 1：阻塞等待结果（不推荐，失去了异步的意义）
        String result = future.get(10, TimeUnit.SECONDS);

        // 方式 2：注册回调（推荐）
        future.thenAccept(r -> System.out.println("报表结果：" + r))
              .exceptionally(ex -> {
                  System.err.println("报表生成失败：" + ex.getMessage());
                  return null;
              });

        // 方式 3：组合多个异步任务
        CompletableFuture<String> report1 = reportService.generateReport("A");
        CompletableFuture<String> report2 = reportService.generateReport("B");
        CompletableFuture.allOf(report1, report2)
            .thenRun(() -> System.out.println("所有报表生成完成"));
    }
}

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (throwable, method, params) -> {
            // 记录日志、发送告警
            log.error("异步方法执行异常 - 方法：{}，参数：{}，异常：{}",
                method.getName(), Arrays.toString(params), throwable.getMessage(), throwable);
        };
    }
}

@Async
public CompletableFuture<String> asyncTaskWithException() {
    try {
        // 业务逻辑
        return CompletableFuture.completedFuture("成功");
    } catch (Exception e) {
        // 通过 CompletableFuture 传递异常
        return CompletableFuture.failedFuture(e);
    }
}

// 调用方
reportService.asyncTaskWithException()
    .exceptionally(ex -> {
        log.error("异步任务失败", ex);
        return "默认值";
    });

@Service
public class UserService {

    // 错误：同类内部调用，@Async 不生效
    public void register(User user) {
        saveUser(user);       // 同步执行！不走代理
        sendWelcomeEmail(user.getEmail());  // 同步执行！不走代理
    }

    @Async
    public void sendWelcomeEmail(String email) {
        // 本应异步执行，但同类调用导致同步执行
    }

    // 解决方案 1：注入自身（推荐）
    @Autowired
    @Lazy  // 避免循环依赖
    private UserService self;

    public void register(User user) {
        saveUser(user);
        self.sendWelcomeEmail(user.getEmail());  // 通过代理调用，异步生效
    }

    // 解决方案 2：拆分到不同类（更推荐，职责更清晰）
}

@Service
public class BadExample {

    @Async
    private void asyncPrivate() { }   // 不生效！CGLIB 无法代理 private 方法

    @Async
    protected void asyncProtected() { } // 不生效！

    @Async
    public void asyncPublic() { }       // 正确
}

@Component
@Slf4j
public class DataCleanupTask {

    @Autowired
    private OrderRepository orderRepository;
    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 每天凌晨 3 点清理 30 天前的已完成订单
     * 使用分布式锁避免多实例重复执行
     */
    @Scheduled(cron = "0 0 3 * * ?")
    public void cleanExpiredOrders() {
        String lockKey = "task:clean-expired-orders";
        String lockValue = UUID.randomUUID().toString();
        Boolean acquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, Duration.ofMinutes(30));

        if (!Boolean.TRUE.equals(acquired)) {
            log.info("其他实例正在执行清理任务，跳过");
            return;
        }

        try {
            LocalDateTime threshold = LocalDateTime.now().minusDays(30);
            int deleted = orderRepository.deleteByStatusAndCreateTimeBefore(
                OrderStatus.COMPLETED, threshold);
            log.info("清理完成，删除 {} 条过期订单", deleted);
        } finally {
            releaseLock(lockKey, lockValue);
        }
    }

    /**
     * 每小时清理过期的验证码缓存
     * 轻量级任务，无需分布式锁
     */
    @Scheduled(cron = "0 0 * * * ?")
    public void cleanExpiredCaptcha() {
        Set<String> keys = redisTemplate.keys("captcha:*");
        if (keys != null && !keys.isEmpty()) {
            int removed = 0;
            for (String key : keys) {
                Long ttl = redisTemplate.getExpire(key, TimeUnit.SECONDS);
                if (ttl != null && ttl <= 0) {
                    redisTemplate.delete(key);
                    removed++;
                }
            }
            log.info("清理 {} 个过期验证码", removed);
        }
    }

    private void releaseLock(String key, String value) {
        String script = "if redis.call('get',KEYS[1])==ARGV[1] " +
                        "then return redis.call('del',KEYS[1]) " +
                        "else return 0 end";
        redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            List.of(key), value);
    }
}

@Service
@Slf4j
public class EmailService {

    @Autowired
    private JavaMailSender mailSender;

    @Value("${spring.mail.username}")
    private String from;

    /**
     * 异步发送简单邮件
     */
    @Async
    public CompletableFuture<Boolean> sendSimpleEmail(String to, String subject, String text) {
        try {
            SimpleMailMessage message = new SimpleMailMessage();
            message.setFrom(from);
            message.setTo(to);
            message.setSubject(subject);
            message.setText(text);
            mailSender.send(message);
            log.info("邮件发送成功：{}", to);
            return CompletableFuture.completedFuture(true);
        } catch (Exception e) {
            log.error("邮件发送失败：{}，原因：{}", to, e.getMessage());
            return CompletableFuture.failedFuture(e);
        }
    }

    /**
     * 异步发送带附件的邮件
     */
    @Async
    public CompletableFuture<Boolean> sendEmailWithAttachment(String to, String subject,
                                                               String text, File attachment) {
        try {
            MimeMessage message = mailSender.createMimeMessage();
            MimeMessageHelper helper = new MimeMessageHelper(message, true);
            helper.setFrom(from);
            helper.setTo(to);
            helper.setSubject(subject);
            helper.setText(text, true);
            helper.addAttachment(attachment.getName(), attachment);
            mailSender.send(message);
            log.info("带附件邮件发送成功：{}", to);
            return CompletableFuture.completedFuture(true);
        } catch (Exception e) {
            log.error("带附件邮件发送失败：{}", to, e.getMessage());
            return CompletableFuture.failedFuture(e);
        }
    }
}

// 调用方
@Service
public class NotificationService {

    @Autowired
    private EmailService emailService;

    public void notifyUser(User user) {
        // 异步发送，不阻塞主流程
        emailService.sendSimpleEmail(
                user.getEmail(),
                "系统通知",
                "您有一条新消息")
            .exceptionally(ex -> {
                // 发送失败不影响主流程，记录日志即可
                log.warn("通知邮件发送失败，用户：{}", user.getId());
                return false;
            });
    }
}

@Service
@Slf4j
public class ReportService {

    @Autowired
    private OrderRepository orderRepository;
    @Autowired
    private UserRepository userRepository;

    /**
     * 异步生成日报表，返回 CompletableFuture
     */
    @Async("reportExecutor")  // 指定专用线程池
    public CompletableFuture<DailyReport> generateDailyReport(LocalDate date) {
        try {
            log.info("开始生成日报表：{}", date);

            // 并行查询多个数据源
            CompletableFuture<BigDecimal> revenueFuture =
                CompletableFuture.supplyAsync(() ->
                    orderRepository.sumRevenueByDate(date));
            CompletableFuture<Long> orderCountFuture =
                CompletableFuture.supplyAsync(() ->
                    orderRepository.countByDate(date));
            CompletableFuture<Long> newUserFuture =
                CompletableFuture.supplyAsync(() ->
                    userRepository.countByRegisterDate(date));

            // 等待所有查询完成，组装报表
            DailyReport report = new DailyReport();
            report.setDate(date);
            report.setRevenue(revenueFuture.get(30, TimeUnit.SECONDS));
            report.setOrderCount(orderCountFuture.get(30, TimeUnit.SECONDS));
            report.setNewUserCount(newUserFuture.get(30, TimeUnit.SECONDS));

            log.info("日报表生成完成：{}", date);
            return CompletableFuture.completedFuture(report);
        } catch (Exception e) {
            log.error("日报表生成失败：{}", date, e);
            return CompletableFuture.failedFuture(e);
        }
    }
}

// 专用线程池配置
@Configuration
@EnableAsync
public class AsyncExecutorConfig {

    /**
     * 报表生成专用线程池（I/O 密集型，线程数较多）
     */
    @Bean("reportExecutor")
    public Executor reportExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("report-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    /**
     * 通用异步线程池
     */
    @Bean("commonExecutor")
    public Executor commonExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

# application.yml - 一行配置启用虚拟线程
spring:
  threads:
    virtual:
      enabled: true

// 启用虚拟线程后，@Async 自动使用虚拟线程
@Async
public CompletableFuture<String> asyncTask() {
    // 此方法在虚拟线程中执行
    System.out.println("线程：" + Thread.currentThread());
    // 输出：线程：VirtualThread@xxx
    return CompletableFuture.completedFuture("done");
}