diff --git a/src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java b/src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java index 2dae7d13..a00482ab 100644 --- a/src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java +++ b/src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java @@ -3,6 +3,7 @@ package com.gaotao.modules.base.dao; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.gaotao.modules.base.entity.PrintTask; import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Options; import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Update; @@ -16,12 +17,17 @@ import java.util.List; @Mapper public interface PrintTaskDao extends BaseMapper { + /** 仅限制打印队列语句的执行等待，不代表数据库网络通信的硬时限，单位为秒。 */ + int QUEUE_SQL_TIMEOUT_SECONDS = 10; + int ARCHIVE_SQL_TIMEOUT_SECONDS = 60; + /** - * 获取待执行的打印任务（使用READPAST避免锁等待） + * 获取待执行的打印任务（使用READPAST跳过行锁，其他等待由语句超时限制） * * @param limit 获取数量 * @return 待执行任务列表 */ + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) @Select("SELECT TOP ${limit} * FROM print_task WITH (READPAST, READCOMMITTEDLOCK) " + "WHERE task_status = 'PENDING' " + "ORDER BY created_date ASC") @@ -33,6 +39,7 @@ public interface PrintTaskDao extends BaseMapper { * @param limit 获取数量 * @return 站点列表 */ + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) @Select("SELECT TOP ${limit} " + "CASE WHEN site IS NULL OR site = '' THEN 'DEFAULT' ELSE site END AS site " + "FROM print_task WITH (READPAST, READCOMMITTEDLOCK) " + @@ -48,6 +55,7 @@ public interface PrintTaskDao extends BaseMapper { * @param limit 获取数量 * @return 待执行任务列表 */ + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) @Select("SELECT TOP ${limit} * FROM print_task WITH (READPAST, READCOMMITTEDLOCK) " + "WHERE task_status = 'PENDING' " + "AND ((#{site} = 'DEFAULT' AND (site IS NULL OR site = '')) OR site = #{site}) " + @@ -55,12 +63,32 @@ public interface PrintTaskDao extends BaseMapper { List selectPendingTasksBySiteWithReadPast(@Param("site") String site, @Param("limit") int limit); - /** Put tasks abandoned by a crashed server back into the print queue. */ - @Update("UPDATE print_task " + - "SET task_status = 'PENDING', started_date = NULL, completed_date = NULL, error_message = NULL " + + /** 领取任务必须确认更新成功，数据库结果不明确时不得发送打印。 */ + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) + @Update("UPDATE print_task SET task_status = 'PROCESSING', started_date = #{startedDate} " + + "WHERE id = #{taskId} AND task_status = 'PENDING'") + int markAsProcessing(@Param("taskId") Long taskId, @Param("startedDate") Date startedDate); + + /** 只更新仍在执行中的任务，避免覆盖其他终态。 */ + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) + @Update("UPDATE print_task SET task_status = 'SUCCESS', completed_date = #{completedDate}, error_message = NULL " + + "WHERE id = #{taskId} AND task_status = 'PROCESSING'") + int markAsSuccess(@Param("taskId") Long taskId, @Param("completedDate") Date completedDate); + + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) + @Update("UPDATE print_task SET task_status = 'FAILED', error_message = #{errorMessage}, " + + "completed_date = #{completedDate}, retry_count = COALESCE(retry_count, 0) + 1 " + + "WHERE id = #{taskId} AND task_status = 'PROCESSING'") + int markAsFailed(@Param("taskId") Long taskId, @Param("errorMessage") String errorMessage, + @Param("completedDate") Date completedDate); + + /** 遗留任务可能已经出纸，转为失败等待人工核对，不再自动加入打印队列。 */ + @Options(timeout = QUEUE_SQL_TIMEOUT_SECONDS) + @Update("UPDATE print_task SET task_status = 'FAILED', completed_date = #{completedDate}, " + + "error_message = N'执行超时或结果未知，可能已出纸，重试前必须人工核对标签及服务日志' " + "WHERE task_status = 'PROCESSING' " + "AND (started_date IS NULL OR started_date < #{before})") - int resetStaleProcessingTasks(@Param("before") Date before); + int resetStaleProcessingTasks(@Param("before") Date before, @Param("completedDate") Date completedDate); /** * 将已完成的打印任务插入到历史表 diff --git a/src/main/java/com/gaotao/modules/base/service/Impl/PrintTaskServiceImpl.java b/src/main/java/com/gaotao/modules/base/service/Impl/PrintTaskServiceImpl.java index 9ae42b2a..88557a41 100644 --- a/src/main/java/com/gaotao/modules/base/service/Impl/PrintTaskServiceImpl.java +++ b/src/main/java/com/gaotao/modules/base/service/Impl/PrintTaskServiceImpl.java @@ -65,8 +65,7 @@ public class PrintTaskServiceImpl extends ServiceImpl i @Override public List getPendingTasks(int limit) { - // 使用原生SQL + READPAST提示，避免锁等待 - // READPAST: 跳过被其他事务锁定的行，确保不阻塞INSERT操作 + // READPAST只跳过行锁，页锁等等待另由打印队列语句超时限制。 return this.baseMapper.selectPendingTasksWithReadPast(limit); } @@ -83,42 +82,27 @@ public class PrintTaskServiceImpl extends ServiceImpl i } @Override - @Transactional + @Transactional(timeout = PrintTaskDao.QUEUE_SQL_TIMEOUT_SECONDS) public int resetStaleProcessingTasks(Date before) { - return this.baseMapper.resetStaleProcessingTasks(before); + return this.baseMapper.resetStaleProcessingTasks(before, new Date()); } @Override - @Transactional + @Transactional(timeout = PrintTaskDao.QUEUE_SQL_TIMEOUT_SECONDS) public boolean markAsProcessing(Long taskId) { - return this.lambdaUpdate() - .set(PrintTask::getTaskStatus, "PROCESSING") - .set(PrintTask::getStartedDate, new Date()) - .eq(PrintTask::getId, taskId) - .eq(PrintTask::getTaskStatus, "PENDING") // 乐观锁：只更新PENDING状态的任务 - .update(); + return this.baseMapper.markAsProcessing(taskId, new Date()) == 1; } @Override - @Transactional + @Transactional(timeout = PrintTaskDao.QUEUE_SQL_TIMEOUT_SECONDS) public boolean markAsSuccess(Long taskId) { - return this.lambdaUpdate() - .set(PrintTask::getTaskStatus, "SUCCESS") - .set(PrintTask::getCompletedDate, new Date()) - .eq(PrintTask::getId, taskId) - .update(); + return this.baseMapper.markAsSuccess(taskId, new Date()) == 1; } @Override - @Transactional + @Transactional(timeout = PrintTaskDao.QUEUE_SQL_TIMEOUT_SECONDS) public boolean markAsFailed(Long taskId, String errorMessage) { - return this.lambdaUpdate() - .set(PrintTask::getTaskStatus, "FAILED") - .set(PrintTask::getErrorMessage, errorMessage) - .set(PrintTask::getCompletedDate, new Date()) - .setSql("retry_count = retry_count + 1") - .eq(PrintTask::getId, taskId) - .update(); + return this.baseMapper.markAsFailed(taskId, errorMessage, new Date()) == 1; } /** @@ -134,7 +118,7 @@ public class PrintTaskServiceImpl extends ServiceImpl i * @return 归档的任务数量 */ @Override - @Transactional + @Transactional(timeout = PrintTaskDao.ARCHIVE_SQL_TIMEOUT_SECONDS) public int archivePrintTaskHistory() { log.info("开始归档打印任务历史数据，归档天数: {} 天", ARCHIVE_DAYS); diff --git a/src/main/java/com/gaotao/modules/base/service/PrintTaskService.java b/src/main/java/com/gaotao/modules/base/service/PrintTaskService.java index 586fb39d..78e50e54 100644 --- a/src/main/java/com/gaotao/modules/base/service/PrintTaskService.java +++ b/src/main/java/com/gaotao/modules/base/service/PrintTaskService.java @@ -53,7 +53,7 @@ public interface PrintTaskService extends IService { */ List getPendingTasksBySite(String site, int limit); - /** Requeue PROCESSING tasks left behind by a stopped server. */ + /** 将超时遗留的执行中任务转为失败，人工核对后才能重试，不自动重新入队。 */ int resetStaleProcessingTasks(Date before); /** diff --git a/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java b/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java index 26296e75..96912edf 100644 --- a/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java +++ b/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java @@ -16,11 +16,15 @@ 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; @@ -54,7 +58,7 @@ public class PrintTaskScheduler { private static final Gson gson = new Gson(); /** - * 打印调度总开关（未配置时默认开启） + * 打印调度总开关（未配置时默认关闭） */ @Value("${dashboard.print.enabled:false}") private boolean dashboardPushEnabled; @@ -84,16 +88,20 @@ public class PrintTaskScheduler { private int printerConnectTimeoutMs; /** - * 打印机读写超时（毫秒，未配置时默认10000） + * 打印机读取超时（不限制写入，毫秒，未配置时默认10000） */ @Value("${print.scheduler.printer-read-timeout-ms:10000}") private int printerReadTimeoutMs; - /** Maximum age of a PROCESSING task before it is considered abandoned. */ + /** 单次打印发送的总时限，包含全部份数及RFID等待时间。 */ + @Value("${print.scheduler.printer-task-timeout-ms:60000}") + private int printerTaskTimeoutMs = 60000; + + /** 执行中任务超过该时长后视为异常遗留任务。 */ @Value("${print.scheduler.stale-processing-timeout-minutes:30}") private int staleProcessingTimeoutMinutes; - /** How often abandoned PROCESSING tasks are recovered. */ + /** 异常遗留任务的恢复检查间隔。 */ @Value("${print.scheduler.recovery-interval-ms:60000}") private long recoveryIntervalMs; @@ -101,6 +109,8 @@ public class PrintTaskScheduler { private static final AtomicLong LAST_IDLE_LOG_AT = new AtomicLong(0); private final Map siteProcessingFlags = new ConcurrentHashMap<>(); private ExecutorService siteTaskExecutor; + private ScheduledThreadPoolExecutor printerTimeoutExecutor; + private final Set activePrinterSockets = ConcurrentHashMap.newKeySet(); @PostConstruct public void initSiteExecutor() { @@ -122,6 +132,17 @@ public class PrintTaskScheduler { 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); @@ -144,6 +165,10 @@ public class PrintTaskScheduler { @PreDestroy public void shutdownSiteExecutor() { + if (printerTimeoutExecutor != null) { + printerTimeoutExecutor.shutdownNow(); + } + activePrinterSockets.forEach(this::closePrinterSocket); if (siteTaskExecutor != null) { siteTaskExecutor.shutdown(); log.info("打印任务站点线程池已关闭"); @@ -204,11 +229,11 @@ public class PrintTaskScheduler { } } catch (Exception e) { - log.error("打印任务调度器执行异常: {}", e.getMessage(), e); + log.error("打印任务调度器执行异常，阶段=查询待打印站点: {}", e.getMessage(), e); } } - // Recover tasks left in PROCESSING after a server crash or forced restart. + // 遗留任务可能已经出纸，只转为失败待核对，不自动重打。 @Scheduled(fixedDelayString = "${print.scheduler.recovery-interval-ms:60000}", initialDelayString = "${print.scheduler.recovery-initial-delay-ms:5000}", scheduler = "printQueueTaskScheduler") @@ -222,11 +247,11 @@ public class PrintTaskScheduler { Date before = new Date(System.currentTimeMillis() - timeoutMillis); int recovered = printTaskService.resetStaleProcessingTasks(before); if (recovered > 0) { - log.warn("Recovered {} stale PROCESSING print tasks older than {} minutes", + log.warn("已将 {} 个超过 {} 分钟的执行中打印任务转为失败，可能已出纸，重试前必须人工核对", recovered, staleProcessingTimeoutMinutes); } } catch (Exception e) { - log.error("Failed to recover stale PROCESSING print tasks", e); + log.error("打印队列遗留任务处理失败，阶段=超时任务转失败", e); } } @@ -234,6 +259,7 @@ public class PrintTaskScheduler { * 处理指定站点的打印任务（站内串行） */ private void processSiteTasks(String site, AtomicBoolean siteProcessingFlag) { + String stage = "查询站点待打印任务"; try { List pendingTasks = printTaskService.getPendingTasksBySite(site, siteBatchSize); if (pendingTasks.isEmpty()) { @@ -242,11 +268,12 @@ public class PrintTaskScheduler { log.info("【打印队列】site={} 发现 {} 个待执行任务", site, pendingTasks.size()); + stage = "处理站点任务"; for (PrintTask task : pendingTasks) { processTask(task); } } catch (Exception e) { - log.error("【打印队列】site任务执行异常 site={}, error={}", site, e.getMessage(), e); + log.error("【打印队列】退出本轮站点任务 site={}, 阶段={}, error={}", site, stage, e.getMessage(), e); } finally { siteProcessingFlag.set(false); } @@ -259,6 +286,9 @@ public class PrintTaskScheduler { Long taskId = task.getId(); String site = task.getSite(); String printerIp = task.getPrinterIp(); + String stage = "领取任务"; + boolean claimed = false; + boolean updatingStatus = false; try { // 1. 标记为执行中（乐观锁，防止并发重复执行） @@ -268,6 +298,8 @@ public class PrintTaskScheduler { return; } + claimed = true; + stage = "解析标签数据"; log.info("【打印队列】开始执行任务 taskId={}, site={}, 打印机={}, 份数={}", taskId, site, printerIp, task.getCopies()); @@ -278,6 +310,7 @@ public class PrintTaskScheduler { } // 3. 执行打印 + stage = "发送打印数据"; boolean success = executePrint( printerIp, task.getZplCode(), @@ -286,18 +319,36 @@ public class PrintTaskScheduler { labelData ); - // 4. 更新任务状态 + // 状态回写异常时结果可能未知，不能再写失败覆盖已提交的成功结果。 + updatingStatus = true; + stage = success ? "发送完成后回写成功" : "发送失败后回写失败"; if (success) { - printTaskService.markAsSuccess(taskId); + if (!printTaskService.markAsSuccess(taskId)) { + throw new IllegalStateException("成功状态未更新，可能已出纸，需人工核对"); + } log.info("【打印队列】✓ 任务完成 taskId={}, site={}, 打印机={}", taskId, site, printerIp); } else { - printTaskService.markAsFailed(taskId, "打印失败，未返回成功标识"); + if (!printTaskService.markAsFailed(taskId, "打印发送失败或超时，可能已部分出纸，重试前请核对标签及服务日志")) { + throw new IllegalStateException("失败状态未更新，需人工核对"); + } log.error("【打印队列】✗ 任务失败 taskId={}, site={}, 打印机={}", taskId, site, printerIp); } } catch (Exception e) { - log.error("【打印队列】✗ 任务执行异常 taskId={}, site={}, 错误: {}", taskId, site, e.getMessage(), e); - printTaskService.markAsFailed(taskId, "执行异常: " + e.getMessage()); + 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); + } } } @@ -314,45 +365,60 @@ public class PrintTaskScheduler { */ private boolean executePrint(String printerIp, String zplCode, Integer copies, String rfidFlag, Map labelData) { - Socket socket = null; + Socket socket = new Socket(); + ScheduledFuture timeoutFuture = null; + AtomicBoolean timedOut = new AtomicBoolean(false); try { - // 1. 连接打印机 - socket = new Socket(); + 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; - - // 2. 直接发送ZPL代码（已在入队时组装好，包括RFID指令） - log.debug("开始发送ZPL到打印机 {}, 份数: {}, RFID: {}", printerIp, actualCopies, rfidFlag); - + byte[] payload = zplCode.getBytes(StandardCharsets.UTF_8); for (int i = 0; i < actualCopies; i++) { - os.write(zplCode.getBytes(StandardCharsets.UTF_8)); + if (timedOut.get() || socket.isClosed() || Thread.currentThread().isInterrupted()) { + return false; + } + os.write(payload); os.flush(); - - // RFID标签需要等待处理时间 if ("Y".equals(rfidFlag)) { - Thread.sleep(1000); // RFID标签等待1秒 - log.debug("RFID标签打印完成 {} / {}", i + 1, actualCopies); + Thread.sleep(1000); // 保留原有RFID间隔，超时后最多等待本次间隔结束。 } } - - log.debug("✓ ZPL已发送到打印机 {}, 份数: {}", printerIp, actualCopies); + 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={}, error={}", printerIp, e.getMessage(), e); + log.error("发送ZPL失败: printerIp={}, timedOut={}, error={}", printerIp, timedOut.get(), e.getMessage(), e); return false; } finally { - if (socket != null) { - try { - socket.close(); - } catch (Exception e) { - log.warn("关闭socket失败: {}", e.getMessage()); - } - } + 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()); } }