# DbJobWorker.java

    package cloud.wconnect.workperformance.scheduler.database;

    import cloud.wconnect.commonlib.cons.FunctionIdConstant;
    import cloud.wconnect.commonlib.cons.SchedulerCons;
    import cloud.wconnect.commonlib.service.CommonService;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    import cloud.wconnect.workperformance.businesstask.database.*;
    import cloud.wconnect.workperformance.cons.database.DbNotifyTypeEnum;
    import cloud.wconnect.workperformance.entity.database.DbScheduler;
    import cloud.wconnect.workperformance.service.AmazonSqsService;
    import cloud.wconnect.workperformance.service.database.IDbSchedulerService;
    import com.alibaba.fastjson.JSONObject;
    import jakarta.annotation.Resource;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    import software.amazon.awssdk.services.sqs.model.Message;

    import java.text.MessageFormat;
    import java.util.List;
    import java.util.Objects;

    @Slf4j
    @Component
    public class DbJobWorker {

        @Resource
        private AmazonSqsService amazonSqsService;
        @Resource
        private IDbSchedulerService schedulerService;
        @Resource
        private DbPmsManagerFlashLast1DayTask managerFlashLast1DayTask;
        @Resource
        private DbPmsManagerFlashLast2DayTask managerFlashLast2DayTask;
        @Resource
        private DbPmsMarketCodeStatisticsLast1DayTask marketCodeStatisticsLast1DayTask;
        @Resource
        private DbPmsMarketCodeStatisticsLast2DayTask marketCodeStatisticsLast2DayTask;
        @Resource
        private DbPmsHfLast2DayTo365DayTask hfLast2DayTo365DayTask;
        @Resource
        private DbPmsOsLast2DayToLast1DayTask osLast2DayToLast1DayTask;
        @Resource
        private DbPmsMssLast2DayToLast1DayTask mssLast2DayToLast1DayTask;
        @Resource
        private MailInternalService mailInternalService;
        @Resource
        private CommonService commonService;

        @Value("${sqs.queue-url.database_scheduler}")
        private String schedulerUrl;

        /**
         * 消费消息
         *
         * @author zc
         * @date 2025/12/02
         */
        @Scheduled(cron = "${schedule.cron.run-scheduler-job}", zone = "${default.timezone}")
        public void run() {
            List<Message> messages = amazonSqsService.getMessagesBatch(schedulerUrl);
            if (CollectionUtils.isNotEmpty(messages)) {
                for (Message message : messages) {
                    // message为json字符串：{"jobId": 1}, 该jobId 为 scheduler 表的 id，该对应记录保存着任务的详细信息
                    // 警告: 执行任务的具体方法应该使用新的线程，而不是在本线程执行（要确保本线程不会被阻塞或异常退出）

                    DbScheduler schedulerJob = null;
                    try {
                        JSONObject jsonObject = JSONObject.parseObject(message.body());
                        String jobId = (String) jsonObject.get(SchedulerCons.JOB_ID);
                        schedulerJob = schedulerService.getById(jobId);

                        if (Objects.nonNull(schedulerJob)) {
                            if (!commonService.checkPropertyEffectiveAndOpenFunction(schedulerJob.getPropertyId(), FunctionIdConstant.DATABASE)) {
                                return;
                            }

                            // 执行任务
                            executeJob(schedulerJob);

                            // 任务执行次数加1
                            schedulerService.plusRunTimes(jobId);
                        }
                    } catch (Exception e) {
                        log.error("SQS consume error[DbJobWorker] - Message: {}; Exception: {}", message.body(), e.getMessage());
                        if (RateLimitUtils.tryAcquire(this)) {
                            sendAlertEmail(schedulerUrl, message.body(), schedulerJob, e.getMessage());
                        }
                    } finally {
                        // 删除消息
                        amazonSqsService.deleteMessage(schedulerUrl, message.receiptHandle());
                    }
                }
                run();
            }
        }

        /**
         * 执行任务
         *
         * @param schedulerJob 调度任务
         * @author zc
         * @date 2025/12/02
         */
        private void executeJob(DbScheduler schedulerJob) {
            int category = schedulerJob.getCategory();
            if (category == DbNotifyTypeEnum.MISSING_REPORT_MANAGER_FLASH_LAST_1_DAY.getType()) {
                managerFlashLast1DayTask.dataMissingCheck(schedulerJob);
            } else if (category == DbNotifyTypeEnum.MISSING_REPORT_MANAGER_FLASH_LAST_2_DAY.getType()) {
                managerFlashLast2DayTask.dataMissingCheck(schedulerJob);
            } else if (category == DbNotifyTypeEnum.MISSING_REPORT_MARKET_CODE_STATISTICS_LAST_1_DAY.getType()) {
                marketCodeStatisticsLast1DayTask.dataMissingCheck(schedulerJob);
            } else if (category == DbNotifyTypeEnum.MISSING_REPORT_MARKET_CODE_STATISTICS_LAST_2_DAY.getType()) {
                marketCodeStatisticsLast2DayTask.dataMissingCheck(schedulerJob);
            } else if (category == DbNotifyTypeEnum.MISSING_REPORT_HF_LAST_2_DAY_TO_365_DAY.getType()) {
                hfLast2DayTo365DayTask.dataMissingCheck(schedulerJob);
            } else if (category == DbNotifyTypeEnum.MISSING_REPORT_OS_LAST_2_DAY_TO_LAST_1_DAY.getType()) {
                osLast2DayToLast1DayTask.dataMissingCheck(schedulerJob);
            } else if (category == DbNotifyTypeEnum.MISSING_REPORT_MSS_LAST_2_DAY_TO_LAST_1_DAY.getType()) {
                mssLast2DayToLast1DayTask.dataMissingCheck(schedulerJob);
            }
        }

        /**
         * 发送alertEmail
         *
         * @param schedulerUrl 定时任务调度URL
         * @param message      消息
         * @param schedulerJob 定时任务对象
         * @param exception    异常信息
         * @author zc
         * @date 2025/12/02
         */
        public void sendAlertEmail(String schedulerUrl, String message, DbScheduler schedulerJob, String exception) {
            String subject = "[Warning] - process DbJobWorker";
            String mailContent = MessageFormat.format("SQS url:{0}<br/>Message:{1}<br/>SchedulerJob:{2}<br/>Exception:{3}",
                    schedulerUrl, message, Objects.nonNull(schedulerJob) ? schedulerJob.toString() : "", exception);
            mailInternalService.sendAlertEmail(subject, mailContent);
        }
    }