xxl-job 阻塞策略，丢弃后续调度，原理及场景

C18298182575

832人浏览 · 2025-11-21 18:02:02

C18298182575 · 2025-11-21 18:02:02 发布

XXL-Job 阻塞策略：丢弃后续调度

一、原理剖析

1. 核心工作机制

java

// 调度器检查逻辑伪代码
public class XxlJobTrigger {
    
    public void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount) {
        // 1. 查询任务是否正在运行
        JobThread jobThread = JobThreadRepository.loadByJobId(jobId);
        
        // 2. 检查阻塞策略
        if (jobThread != null && jobThread.isRunning()) {
            ExecutorBlockStrategyEnum blockStrategy = loadBlockStrategy(jobId);
            
            switch (blockStrategy) {
                case DISCARD_LATER:
                    // 丢弃后续调度：直接返回，不执行本次触发
                    logger.warn("任务[{}]正在运行，丢弃本次调度", jobId);
                    return;
                case SERIAL_EXECUTION:
                    // 串行执行：加入队列等待
                    jobThread.pushTriggerQueue(triggerParam);
                    break;
                case COVER_EARLY:
                    // 覆盖之前调度：终止当前执行，开始新的
                    jobThread.interrupt();
                    break;
            }
        }
        
        // 3. 如果没有运行或策略允许，开始执行
        if (jobThread == null || !jobThread.isRunning()) {
            startNewExecution(jobId, triggerParam);
        }
    }
}

2. 状态机流转

二、源码深度解析

1. 阻塞策略枚举

java

public enum ExecutorBlockStrategyEnum {
    
    SERIAL_EXECUTION("串行执行"),
    DISCARD_LATER("丢弃后续调度"), 
    COVER_EARLY("覆盖之前调度");
    
    private String title;
    
    // 策略选择逻辑
    public static ExecutorBlockStrategyEnum match(String name, ExecutorBlockStrategyEnum defaultItem) {
        if (name != null) {
            for (ExecutorBlockStrategyEnum item : ExecutorBlockStrategyEnum.values()) {
                if (item.name().equals(name)) {
                    return item;
                }
            }
        }
        return defaultItem;
    }
}

2. 任务线程管理

java

public class JobThread extends Thread {
    
    private int jobId;
    private volatile boolean running = false;
    private LinkedBlockingQueue<TriggerParam> triggerQueue;
    
    @Override
    public void run() {
        while (!toStop) {
            try {
                // 从队列获取触发参数
                TriggerParam triggerParam = triggerQueue.take();
                
                // 设置运行状态
                running = true;
                
                // 执行任务
                executorBiz.run(triggerParam);
                
            } finally {
                running = false;
            }
        }
    }
    
    public boolean isRunning() {
        return running;
    }
}

三、适用场景分析

✅ 适合使用"丢弃后续调度"的场景

1. 高频数据采集任务

java

@XxlJob("dataCollector")
public void dataCollector() {
    // 场景：每10秒采集一次传感器数据
    // 如果前一次采集还在进行，说明系统繁忙
    // 丢弃本次调度避免堆积
    SensorData data = sensorService.collectData();
    dataProcessor.process(data);
}

理由：数据具有时效性，旧数据采集未完成时，新数据可能已过时

2. 实时监控告警

java

@XxlJob("systemMonitor")
public void systemMonitor() {
    // 监控系统状态，生成告警
    // 如果上一次监控还在分析，直接丢弃避免重复告警
    SystemStatus status = monitorService.getSystemStatus();
    if (status.isAbnormal()) {
        alertService.sendAlert(status);
    }
}

理由：监控告警不需要堆积，保持最新状态即可

3. 缓存刷新任务

java

@XxlJob("cacheRefresh")
public void cacheRefresh() {
    // 刷新用户信息缓存
    // 如果正在刷新，丢弃后续请求，避免缓存穿透
    userCache.refreshAll();
}

理由：缓存刷新操作较重，连续刷新无意义

4. 报表生成任务

java

@XxlJob("dailyReport")
public void generateDailyReport() {
    // 生成每日报表，耗时较长
    // 如果正在生成，丢弃新的生成请求
    Report report = reportService.generateDailyReport(LocalDate.now());
    emailService.sendReport(report);
}

理由：同一份报表不需要重复生成

❌ 不适合使用的场景

1. 订单处理任务

java

@XxlJob("orderProcessor")
public void processOrders() {
    // 错误用法：订单处理必须保证每个订单都被处理
    // 使用"串行执行"更合适
    List<Order> orders = orderService.fetchPendingOrders();
    orders.forEach(order -> orderService.process(order));
}

2. 财务对账任务

java

@XxlJob("financialReconciliation") 
public void financialReconciliation() {
    // 错误用法：财务数据必须完整处理
    // 使用"串行执行"确保数据一致性
    reconciliationService.processDailyTransactions();
}

四、生产环境配置

1. 管理端配置

properties

# 在XXL-Job管理界面配置
任务名称: 用户行为数据采集
任务描述: 采集用户点击流数据
调度类型: CRON
Cron表达式: 0/30 * * * * ?
阻塞处理策略: 丢弃后续调度

2. 代码注解配置

java

@XxlJob(value = "userBehaviorCollector")
public ReturnT<String> userBehaviorCollector(String param) {
    try {
        // 业务逻辑
        userBehaviorService.collectAndAnalyze();
        return ReturnT.SUCCESS;
    } catch (Exception e) {
        return ReturnT.FAIL;
    }
}

五、监控与告警

1. 丢弃次数监控

java

@Component
public class BlockStrategyMonitor {
    
    @XxlJob("blockStrategyMonitor")
    public void monitor() {
        // 监控丢弃策略的任务
        List<XxlJobLog> discardedJobs = xxlJobLogMapper.selectDiscardedJobs();
        
        if (!discardedJobs.isEmpty()) {
            // 发送告警
            alertService.sendAlert("发现任务丢弃", discardedJobs);
        }
    }
}

2. 日志分析

sql

-- 查询被丢弃的调度记录
SELECT * FROM xxl_job_log 
WHERE trigger_code = 500 
AND handle_code = 0 
AND trigger_msg LIKE '%丢弃%';

六、最佳实践

1. 任务设计原则

java

public class DiscardStrategyBestPractice {
    
    // 好任务：幂等、可丢弃、非关键
    @XxlJob("goodExample")
    public void goodExample() {
        // 特点：执行失败不影响业务连续性
        cacheService.refresh();
        statisticService.calculate();
        logService.archive();
    }
    
    // 坏任务：非幂等、关键业务
    @XxlJob("badExample") 
    public void badExample() {
        // 特点：必须保证每次执行
        paymentService.settleAccounts();
        orderService.deliverGoods();
    }
}

2. 超时时间配置

java

@XxlJob("timeoutAwareJob")
public ReturnT<String> timeoutAwareJob(String param) {
    // 设置合理的超时时间，避免长时间阻塞
    XxlJobHelper.handleTimeout(300000); // 5分钟超时
    
    // 业务逻辑
    heavyCalculationService.calculate();
    
    return ReturnT.SUCCESS;
}