Spring Boot微服务动态Cron调度多批量邮件需求实现问询

核心思路 #

基于Spring原生的TaskScheduler和ScheduledFuture实现任务的动态注册、更新与删除，结合数据库存储Cron表达式，配合前端接口完成配置修改后的实时生效。

具体实现步骤 #

1. 定义任务配置实体与数据库表 #

先创建存储任务配置的实体类，对应数据库表结构：

@Entity
@Table(name = "email_task_config")
public class EmailTaskConfig {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    // 任务唯一标识，用于区分不同批量邮件任务，比如"user_notify_batch"
    private String taskKey;
    // 调度用Cron表达式
    private String cronExpression;
    // 任务描述
    private String description;
    // 是否启用该任务
    private Boolean enabled;

    // getter、setter方法省略
}

2. 封装批量邮件执行逻辑 #

把不同类型的批量邮件逻辑抽成独立的可执行任务，通过taskKey区分不同任务：

public class BatchEmailTask implements Runnable {
    private final String taskKey;

    // 构造器传入任务标识，绑定对应执行逻辑
    public BatchEmailTask(String taskKey) {
        this.taskKey = taskKey;
    }

    @Override
    public void run() {
        switch (taskKey) {
            case "user_notify_batch":
                sendUserNotifyEmails();
                break;
            case "marketing_email_batch":
                sendMarketingEmails();
                break;
            // 扩展其他批量邮件任务分支
        }
    }

    private void sendUserNotifyEmails() {
        // 具体逻辑：查询待通知用户、构造邮件内容、调用邮件服务发送
    }

    private void sendMarketingEmails() {
        // 营销批量邮件发送逻辑
    }
}

3. 实现任务调度管理器 #

核心类负责加载数据库任务、调度任务、更新任务：

@Component
public class TaskSchedulerManager implements ApplicationListener<ContextRefreshedEvent> {
    @Autowired
    private TaskScheduler taskScheduler;
    @Autowired
    private EmailTaskConfigRepository configRepository;

    // 存储已调度任务，key为taskKey，value为调度返回的Future对象
    private final Map<String, ScheduledFuture<?>> scheduledTasks = new ConcurrentHashMap<>();

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        // 项目启动时自动加载所有启用的任务
        loadAllEnabledTasks();
    }

    // 加载所有启用的任务
    public void loadAllEnabledTasks() {
        List<EmailTaskConfig> enabledConfigs = configRepository.findByEnabled(true);
        for (EmailTaskConfig config : enabledConfigs) {
            scheduleSingleTask(config);
        }
    }

    // 调度单个任务
    public void scheduleSingleTask(EmailTaskConfig config) {
        // 先取消同key的已存在任务，避免重复调度
        cancelExistingTask(config.getTaskKey());
        // 创建任务实例
        BatchEmailTask task = new BatchEmailTask(config.getTaskKey());
        // 基于Cron表达式调度任务
        ScheduledFuture<?> future = taskScheduler.schedule(task, new CronTrigger(config.getCronExpression()));
        scheduledTasks.put(config.getTaskKey(), future);
    }

    // 更新任务（前端修改配置后调用此方法）
    public void updateTask(Long configId) {
        EmailTaskConfig config = configRepository.findById(configId)
                .orElseThrow(() -> new RuntimeException("任务配置不存在"));
        if (config.getEnabled()) {
            scheduleSingleTask(config);
        } else {
            cancelExistingTask(config.getTaskKey());
        }
    }

    // 取消已存在的任务
    private void cancelExistingTask(String taskKey) {
        ScheduledFuture<?> future = scheduledTasks.get(taskKey);
        if (future != null && !future.isCancelled()) {
            future.cancel(false);
            scheduledTasks.remove(taskKey);
        }
    }
}

4. 提供前端交互接口 #

编写Controller接口，供前端修改任务配置并触发调度更新：

@RestController
@RequestMapping("/api/email-tasks")
public class EmailTaskController {
    @Autowired
    private EmailTaskConfigRepository configRepository;
    @Autowired
    private TaskSchedulerManager schedulerManager;

    // 修改任务Cron表达式或启用状态
    @PutMapping("/{id}")
    public ResponseEntity<Void> updateTaskConfig(@PathVariable Long id, @RequestBody EmailTaskConfig updateInfo) {
        EmailTaskConfig existingConfig = configRepository.findById(id)
                .orElseThrow(() -> new RuntimeException("任务不存在"));
        // 更新Cron表达式
        if (updateInfo.getCronExpression() != null) {
            // 提前校验Cron合法性
            if (!CronExpression.isValidExpression(updateInfo.getCronExpression())) {
                throw new RuntimeException("无效的Cron表达式");
            }
            existingConfig.setCronExpression(updateInfo.getCronExpression());
        }
        // 更新启用状态
        if (updateInfo.getEnabled() != null) {
            existingConfig.setEnabled(updateInfo.getEnabled());
        }
        configRepository.save(existingConfig);
        // 触发调度更新
        schedulerManager.updateTask(id);
        return ResponseEntity.ok().build();
    }

    // 获取所有任务配置
    @GetMapping
    public ResponseEntity<List<EmailTaskConfig>> getAllTaskConfigs() {
        return ResponseEntity.ok(configRepository.findAll());
    }
}

5. 关键注意事项 #

• 用ConcurrentHashMap存储已调度任务，避免并发操作时的线程安全问题
• 更新任务必须先取消旧任务再调度新任务，防止同一任务重复执行
• 前端传入Cron表达式时，必须做合法性校验，避免无效配置导致调度失败
• 批量邮件发送逻辑建议异步执行，避免阻塞调度线程影响其他任务

内容的提问来源于stack exchange，提问作者Aditya Wadhawan