wms-java/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java

package com.gaotao.modules.base.task;

import com.gaotao.modules.base.entity.PrintTask;
import com.gaotao.modules.base.service.PrintTaskService;
import com.google.gson.Gson;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.nio.charset.StandardCharsets;

/**
 * 打印任务定时调度器
 *
 * <p><b>功能说明：</b></p>
 * <ul>
 *   <li>每2秒扫描一次打印任务队列</li>
 *   <li>按站点（site）拆分任务，不同站点并发处理</li>
 *   <li>同一站点内按FIFO顺序（created_date）取出PENDING状态的任务</li>
 *   <li>执行打印任务并更新状态</li>
 *   <li>失败任务记录错误信息</li>
 * </ul>
 *
 * <p><b>并发控制：</b></p>
 * <ul>
 *   <li>使用数据库乐观锁（markAsProcessing时检查状态）</li>
 *   <li>确保同一任务不会被重复执行</li>
 * </ul>
 */
@Slf4j
@Component
public class PrintTaskScheduler {

    @Autowired
    private PrintTaskService printTaskService;

    private static final Gson gson = new Gson();

    /**
     * 打印调度总开关（未配置时默认关闭）
     */
    @Value("${dashboard.print.enabled:false}")
    private boolean dashboardPushEnabled;

    /**
     * 每轮调度最多扫描的站点数量（未配置时默认20）
     */
    @Value("${print.scheduler.site.max-sites-per-round:20}")
    private int maxSitesPerRound;

    /**
     * 每个站点每轮最多处理的任务数量（未配置时默认10）
     */
    @Value("${print.scheduler.site.batch-size:10}")
    private int siteBatchSize;

    /**
     * 站点并发线程池大小（未配置时默认8）
     */
    @Value("${print.scheduler.site.thread-pool-size:8}")
    private int siteThreadPoolSize;

    /**
     * 打印机连接超时（毫秒，未配置时默认3000）
     */
    @Value("${print.scheduler.printer-connect-timeout-ms:3000}")
    private int printerConnectTimeoutMs;

    /**
     * 打印机读取超时（不限制写入，毫秒，未配置时默认10000）
     */
    @Value("${print.scheduler.printer-read-timeout-ms:10000}")
    private int printerReadTimeoutMs;

    /** 单次打印发送的总时限，包含全部份数及RFID等待时间。 */
    @Value("${print.scheduler.printer-task-timeout-ms:60000}")
    private int printerTaskTimeoutMs = 60000;

    /** 执行中任务超过该时长后视为异常遗留任务。 */
    @Value("${print.scheduler.stale-processing-timeout-minutes:30}")
    private int staleProcessingTimeoutMinutes;

    /** 异常遗留任务的恢复检查间隔。 */
    @Value("${print.scheduler.recovery-interval-ms:60000}")
    private long recoveryIntervalMs;

    private static final AtomicInteger THREAD_INDEX = new AtomicInteger(1);
    private static final AtomicLong LAST_IDLE_LOG_AT = new AtomicLong(0);
    private final Map<String, AtomicBoolean> siteProcessingFlags = new ConcurrentHashMap<>();
    private ExecutorService siteTaskExecutor;
    private ScheduledThreadPoolExecutor printerTimeoutExecutor;
    private final Set<Socket> activePrinterSockets = ConcurrentHashMap.newKeySet();

    @PostConstruct
    public void initSiteExecutor() {
        if (maxSitesPerRound <= 0) {
            log.warn("Invalid print.scheduler.site.max-sites-per-round={}, using 20", maxSitesPerRound);
            maxSitesPerRound = 20;
        }
        if (siteBatchSize <= 0) {
            log.warn("Invalid print.scheduler.site.batch-size={}, using 10", siteBatchSize);
            siteBatchSize = 10;
        }
        if (staleProcessingTimeoutMinutes <= 0) {
            log.warn("Invalid print.scheduler.stale-processing-timeout-minutes={}, using 30",
                    staleProcessingTimeoutMinutes);
            staleProcessingTimeoutMinutes = 30;
        }
        if (recoveryIntervalMs < 1000) {
            log.warn("Invalid print.scheduler.recovery-interval-ms={}, using 60000", recoveryIntervalMs);
            recoveryIntervalMs = 60000;
        }

        if (printerTaskTimeoutMs <= 0) {
            log.warn("打印任务超时配置无效，使用60000毫秒，原配置={}", printerTaskTimeoutMs);
            printerTaskTimeoutMs = 60000;
        }
        printerTimeoutExecutor = new ScheduledThreadPoolExecutor(1, runnable -> {
            Thread thread = new Thread(runnable, "print-socket-timeout");
            thread.setDaemon(true);
            return thread;
        });
        printerTimeoutExecutor.setRemoveOnCancelPolicy(true);

        int actualThreadPoolSize = Math.max(1, siteThreadPoolSize);
        if (actualThreadPoolSize != siteThreadPoolSize) {
            log.warn("打印站点线程池配置无效，使用默认最小值1，原配置={}", siteThreadPoolSize);
        }

        ThreadFactory threadFactory = runnable -> {
            Thread thread = new Thread(runnable);
            thread.setName("print-site-worker-" + THREAD_INDEX.getAndIncrement());
            return thread;
        };
        siteTaskExecutor = Executors.newFixedThreadPool(actualThreadPoolSize, threadFactory);
        if (dashboardPushEnabled) {
            log.info("Print task scheduler config: enabled={}, maxSites={}, batchSize={}, siteWorkers={}, " +
                            "staleTimeoutMinutes={}, recoveryIntervalMs={}",
                    dashboardPushEnabled, maxSitesPerRound, siteBatchSize, actualThreadPoolSize,
                    staleProcessingTimeoutMinutes, recoveryIntervalMs);
        }
        log.info("打印任务站点线程池初始化完成，线程数={}", actualThreadPoolSize);
    }

    @PreDestroy
    public void shutdownSiteExecutor() {
        if (printerTimeoutExecutor != null) {
            printerTimeoutExecutor.shutdownNow();
        }
        activePrinterSockets.forEach(this::closePrinterSocket);
        if (siteTaskExecutor != null) {
            siteTaskExecutor.shutdown();
            log.info("打印任务站点线程池已关闭");
        }
    }

    // 每2秒执行一次
    @Scheduled(fixedDelayString = "${print.scheduler.poll-interval-ms:2000}",
            scheduler = "printQueueTaskScheduler")
    public void processPrintTasks() {
        // 检查定时任务开关
        if (!dashboardPushEnabled) {
            return;
        } else {
            log.debug("Print task scheduler tick: enabled={}, thread={}",
                    dashboardPushEnabled, Thread.currentThread().getName());
        }
        if (siteTaskExecutor == null || siteTaskExecutor.isShutdown()) {
            log.warn("打印任务站点线程池不可用，跳过本轮调度");
            return;
        }
        try {
            // 1. 获取有待执行任务的站点列表
            List<String> pendingSites = printTaskService.getPendingSites(maxSitesPerRound);
            if (pendingSites.isEmpty()) {
                long now = System.currentTimeMillis();
                long lastLog = LAST_IDLE_LOG_AT.get();
                if (now - lastLog >= 30000 && LAST_IDLE_LOG_AT.compareAndSet(lastLog, now)) {
                    log.info("Print task scheduler is running; no PENDING sites found");
                }
                return; // 没有待处理任务
            }

            log.info("【打印队列】发现 {} 个待执行站点", pendingSites.size());

            // 2. 按站点并发处理，不同站点互不影响
            for (String site : pendingSites) {
                if (site == null || site.trim().isEmpty()) {
                    continue;
                }

                AtomicBoolean siteProcessingFlag = siteProcessingFlags
                        .computeIfAbsent(site, key -> new AtomicBoolean(false));

                // 同一个site只允许一个worker在执行，避免站内并发
                if (!siteProcessingFlag.compareAndSet(false, true)) {
                    log.debug("【打印队列】site={} 正在执行中，跳过本轮", site);
                    continue;
                }

                try {
                    siteTaskExecutor.submit(() -> processSiteTasks(site, siteProcessingFlag));
                } catch (Exception submitEx) {
                    siteProcessingFlag.set(false);
                    log.error("【打印队列】提交site任务失败 site={}, error={}",
                            site, submitEx.getMessage(), submitEx);
                }
            }

        } catch (Exception e) {
            log.error("打印任务调度器执行异常，阶段=查询待打印站点: {}", e.getMessage(), e);
        }
    }

    // 遗留任务可能已经出纸，只转为失败待核对，不自动重打。
    @Scheduled(fixedDelayString = "${print.scheduler.recovery-interval-ms:60000}",
            initialDelayString = "${print.scheduler.recovery-initial-delay-ms:5000}",
            scheduler = "printQueueTaskScheduler")
    public void recoverStaleProcessingTasks() {
        if (!dashboardPushEnabled) {
            return;
        }

        try {
            long timeoutMillis = staleProcessingTimeoutMinutes * 60_000L;
            Date before = new Date(System.currentTimeMillis() - timeoutMillis);
            int recovered = printTaskService.resetStaleProcessingTasks(before);
            if (recovered > 0) {
                log.warn("已将 {} 个超过 {} 分钟的执行中打印任务转为失败，可能已出纸，重试前必须人工核对",
                        recovered, staleProcessingTimeoutMinutes);
            }
        } catch (Exception e) {
            log.error("打印队列遗留任务处理失败，阶段=超时任务转失败", e);
        }
    }

    /**
     * 处理指定站点的打印任务（站内串行）
     */
    private void processSiteTasks(String site, AtomicBoolean siteProcessingFlag) {
        String stage = "查询站点待打印任务";
        try {
            List<PrintTask> pendingTasks = printTaskService.getPendingTasksBySite(site, siteBatchSize);
            if (pendingTasks.isEmpty()) {
                return;
            }

            log.info("【打印队列】site={} 发现 {} 个待执行任务", site, pendingTasks.size());

            stage = "处理站点任务";
            for (PrintTask task : pendingTasks) {
                processTask(task);
            }
        } catch (Exception e) {
            log.error("【打印队列】退出本轮站点任务 site={}, 阶段={}, error={}", site, stage, e.getMessage(), e);
        } finally {
            siteProcessingFlag.set(false);
        }
    }

    /**
     * 处理单个打印任务
     */
    private void processTask(PrintTask task) {
        Long taskId = task.getId();
        String site = task.getSite();
        String printerIp = task.getPrinterIp();
        String stage = "领取任务";
        boolean claimed = false;
        boolean updatingStatus = false;

        try {
            // 1. 标记为执行中（乐观锁，防止并发重复执行）
            boolean marked = printTaskService.markAsProcessing(taskId);
            if (!marked) {
                log.debug("任务 {} 已被其他线程处理，跳过", taskId);
                return;
            }

            claimed = true;
            stage = "解析标签数据";
            log.info("【打印队列】开始执行任务 taskId={}, site={}, 打印机={}, 份数={}",
                    taskId, site, printerIp, task.getCopies());

            // 2. 解析标签数据
            Map<String, Object> labelData = null;
            if (task.getLabelData() != null && !task.getLabelData().trim().isEmpty()) {
                labelData = gson.fromJson(task.getLabelData(), Map.class);
            }

            // 3. 执行打印
            stage = "发送打印数据";
            boolean success = executePrint(
                printerIp,
                task.getZplCode(),
                task.getCopies(),
                task.getRfidFlag(),
                labelData
            );

            // 状态回写异常时结果可能未知，不能再写失败覆盖已提交的成功结果。
            updatingStatus = true;
            stage = success ? "发送完成后回写成功" : "发送失败后回写失败";
            if (success) {
                if (!printTaskService.markAsSuccess(taskId)) {
                    throw new IllegalStateException("成功状态未更新，可能已出纸，需人工核对");
                }
                log.info("【打印队列】✓ 任务完成 taskId={}, site={}, 打印机={}", taskId, site, printerIp);
            } else {
                if (!printTaskService.markAsFailed(taskId, "打印发送失败或超时，可能已部分出纸，重试前请核对标签及服务日志")) {
                    throw new IllegalStateException("失败状态未更新，需人工核对");
                }
                log.error("【打印队列】✗ 任务失败 taskId={}, site={}, 打印机={}", taskId, site, printerIp);
            }

        } catch (Exception e) {
            log.error("【打印队列】任务异常 taskId={}, site={}, 阶段={}, 错误={}",
                    taskId, site, stage, e.getMessage(), e);
            // 领取结果不确定时不打印、不改失败；回写结果不确定时停止本批次，由遗留任务检查兜底。
            if (!claimed || updatingStatus) {
                throw new IllegalStateException("数据库操作未确认，停止当前站点批次，taskId=" + taskId, e);
            }
            try {
                if (!printTaskService.markAsFailed(taskId, "执行异常，重试前请核对: " + e.getMessage())) {
                    throw new IllegalStateException("异常任务的失败状态未更新");
                }
            } catch (Exception updateEx) {
                log.error("【打印队列】异常后回写失败 taskId={}, site={}, 阶段=异常回写失败", taskId, site, updateEx);
                throw new IllegalStateException("失败状态回写未确认，停止当前站点批次，taskId=" + taskId, updateEx);
            }
        }
    }

    /**
     * 执行打印（发送ZPL到打印机）
     * 注意：ZPL代码已在入队时组装好（包括RFID指令），这里只需要直接发送
     *
     * @param printerIp 打印机IP
     * @param zplCode ZPL代码（已包含RFID指令）
     * @param copies 打印份数
     * @param rfidFlag RFID标识（用于日志）
     * @param labelData 标签数据（未使用，保留以备后续扩展）
     * @return 是否成功
     */
    private boolean executePrint(String printerIp, String zplCode, Integer copies,
                                 String rfidFlag, Map<String, Object> labelData) {
        Socket socket = new Socket();
        ScheduledFuture<?> timeoutFuture = null;
        AtomicBoolean timedOut = new AtomicBoolean(false);
        try {
            activePrinterSockets.add(socket);
            // 保持原站点线程串行发送，只用独立计时器关连接，不另开线程接管任务。
            // 先注册超时再连接，避免写入阻塞；取消成功任务的计时器，防止队列累积。
            timeoutFuture = printerTimeoutExecutor.schedule(() -> {
                timedOut.set(true);
                closePrinterSocket(socket);
                log.error("打印发送超过{}毫秒，已关闭连接，可能部分出纸，重试前请核对: printerIp={}",
                        printerTaskTimeoutMs, printerIp);
            }, printerTaskTimeoutMs, TimeUnit.MILLISECONDS);
            if (zplCode == null || zplCode.isBlank()) {
                throw new IllegalArgumentException("打印内容为空");
            }
            socket.connect(new InetSocketAddress(printerIp, 9100), printerConnectTimeoutMs);
            socket.setSoTimeout(printerReadTimeoutMs);
            OutputStream os = socket.getOutputStream();
            int actualCopies = copies != null && copies > 0 ? copies : 1;
            byte[] payload = zplCode.getBytes(StandardCharsets.UTF_8);
            for (int i = 0; i < actualCopies; i++) {
                if (timedOut.get() || socket.isClosed() || Thread.currentThread().isInterrupted()) {
                    return false;
                }
                os.write(payload);
                os.flush();
                if ("Y".equals(rfidFlag)) {
                    Thread.sleep(1000); // 保留原有RFID间隔，超时后最多等待本次间隔结束。
                }
            }
            if (timedOut.get() || socket.isClosed()) return false;
            log.debug("ZPL已发送到打印机 {}, 份数: {}（不代表实际出纸确认）", printerIp, actualCopies);
            return true;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("打印发送被中断: printerIp={}", printerIp);
            return false;
        } catch (Exception e) {
            log.error("发送ZPL失败: printerIp={}, timedOut={}, error={}", printerIp, timedOut.get(), e.getMessage(), e);
            return false;
        } finally {
            if (timeoutFuture != null) timeoutFuture.cancel(false);
            closePrinterSocket(socket);
            activePrinterSockets.remove(socket);
        }
    }

    /** 关闭连接会解除阻塞的网络写入，让原工作线程正常退出并释放站点占用。 */
    private void closePrinterSocket(Socket socket) {
        try {
            socket.close();
        } catch (Exception e) {
            log.warn("关闭打印机连接失败: {}", e.getMessage());
        }
    }

    /**
     * 归档打印任务历史数据
     *
     * <p><b>功能说明：</b></p>
     * <ul>
     *   <li>每天凌晨5点自动执行</li>
     *   <li>将已完成的打印任务备份到 print_task_history 表</li>
     *   <li>清理10天之前的历史数据</li>
     * </ul>
     *
     * <p><b>执行时间：</b>每天凌晨5点（固定）</p>
     * <p><b>归档规则：</b>归档10天前的已完成任务（固定）</p>
     */
    @Scheduled(cron = "0 0 5 * * ?", scheduler = "printQueueTaskScheduler")
    public void archivePrintTaskHistory() {
        // 检查定时任务开关
        if (!dashboardPushEnabled) {
            return;
        }

        log.info("=== 开始归档打印任务历史数据 ===");

        try {
            // 调用Service层执行归档操作
            int archivedCount = printTaskService.archivePrintTaskHistory();

            log.info("=== 打印任务历史数据归档完成：归档任务数量={} ===", archivedCount);

        } catch (Exception e) {
            log.error("=== 打印任务历史数据归档失败 ===", e);
        }
    }
}