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 1fe9e2c4..c85a0a8b 100644 --- a/src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java +++ b/src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java @@ -25,6 +25,34 @@ public interface PrintTaskDao extends BaseMapper { "ORDER BY created_date ASC") List selectPendingTasksWithReadPast(@Param("limit") int limit); + /** + * 获取有待执行任务的站点列表（按站点最早任务时间排序） + * + * @param limit 获取数量 + * @return 站点列表 + */ + @Select("SELECT TOP ${limit} " + + "CASE WHEN site IS NULL OR site = '' THEN 'DEFAULT' ELSE site END AS site " + + "FROM print_task WITH (READPAST) " + + "WHERE task_status = 'PENDING' " + + "GROUP BY CASE WHEN site IS NULL OR site = '' THEN 'DEFAULT' ELSE site END " + + "ORDER BY MIN(created_date) ASC") + List selectPendingSitesWithReadPast(@Param("limit") int limit); + + /** + * 获取指定站点待执行任务（按FIFO） + * + * @param site 站点编码 + * @param limit 获取数量 + * @return 待执行任务列表 + */ + @Select("SELECT TOP ${limit} * FROM print_task WITH (READPAST) " + + "WHERE task_status = 'PENDING' " + + "AND ((#{site} = 'DEFAULT' AND (site IS NULL OR site = '')) OR site = #{site}) " + + "ORDER BY created_date ASC") + List selectPendingTasksBySiteWithReadPast(@Param("site") String site, + @Param("limit") int limit); + /** * 将已完成的打印任务插入到历史表 * 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 abd98ba6..050bd404 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,6 +65,18 @@ public class PrintTaskServiceImpl extends ServiceImpl i return this.baseMapper.selectPendingTasksWithReadPast(limit); } + @Override + public List getPendingSites(int limit) { + // 查询有待执行任务的站点，按最早任务时间排序 + return this.baseMapper.selectPendingSitesWithReadPast(limit); + } + + @Override + public List getPendingTasksBySite(String site, int limit) { + // 按站点查询待执行任务，确保站内FIFO顺序 + return this.baseMapper.selectPendingTasksBySiteWithReadPast(site, limit); + } + @Override @Transactional public boolean markAsProcessing(Long taskId) { 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 62ce694e..803dae28 100644 --- a/src/main/java/com/gaotao/modules/base/service/PrintTaskService.java +++ b/src/main/java/com/gaotao/modules/base/service/PrintTaskService.java @@ -35,6 +35,23 @@ public interface PrintTaskService extends IService { */ List getPendingTasks(int limit); + /** + * 获取有待执行任务的站点列表（按最早任务时间排序） + * + * @param limit 获取站点数量 + * @return 待执行站点列表 + */ + List getPendingSites(int limit); + + /** + * 获取指定站点待执行的打印任务（FIFO顺序） + * + * @param site 站点编码 + * @param limit 获取任务数量 + * @return 待执行任务列表 + */ + List getPendingTasksBySite(String site, int limit); + /** * 更新任务为执行中 * 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 cce9cc4c..ff556785 100644 --- a/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java +++ b/src/main/java/com/gaotao/modules/base/task/PrintTaskScheduler.java @@ -3,6 +3,8 @@ 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; @@ -10,17 +12,26 @@ 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.List; 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.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.nio.charset.StandardCharsets; /** * 打印任务定时调度器 * *

功能说明：

*
    - *
  • 每秒扫描一次打印任务队列
    • - *
  • 按FIFO顺序（created_date）取出PENDING状态的任务
    • + *
  • 每2秒扫描一次打印任务队列
    • + *
  • 按站点（site）拆分任务，不同站点并发处理
    • + *
  • 同一站点内按FIFO顺序（created_date）取出PENDING状态的任务
    • *
  • 执行打印任务并更新状态
    • *
  • 失败任务记录错误信息
    • *
@@ -40,9 +51,70 @@ public class PrintTaskScheduler { private static final Gson gson = new Gson(); + /** + * 打印调度总开关（未配置时默认开启） + */ @Value("${dashboard.push.enabled:true}") 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; + + private static final AtomicInteger THREAD_INDEX = new AtomicInteger(1); + private final Map siteProcessingFlags = new ConcurrentHashMap<>(); + private ExecutorService siteTaskExecutor; + + @PostConstruct + public void initSiteExecutor() { + 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); + log.info("打印任务站点线程池初始化完成，线程数={}", actualThreadPoolSize); + } + + @PreDestroy + public void shutdownSiteExecutor() { + if (siteTaskExecutor != null) { + siteTaskExecutor.shutdown(); + log.info("打印任务站点线程池已关闭"); + } + } + // 每2秒执行一次 @Scheduled(fixedDelay = 2000) public void processPrintTasks() { @@ -50,23 +122,67 @@ public class PrintTaskScheduler { if (!dashboardPushEnabled) { return; } + if (siteTaskExecutor == null || siteTaskExecutor.isShutdown()) { + log.warn("打印任务站点线程池不可用，跳过本轮调度"); + return; + } try { - // 1. 获取待执行的任务（每次处理10个） - List pendingTasks = printTaskService.getPendingTasks(10); + // 1. 获取有待执行任务的站点列表 + List pendingSites = printTaskService.getPendingSites(maxSitesPerRound); + if (pendingSites.isEmpty()) { + 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); + } + } + + /** + * 处理指定站点的打印任务（站内串行） + */ + private void processSiteTasks(String site, AtomicBoolean siteProcessingFlag) { + try { + List pendingTasks = printTaskService.getPendingTasksBySite(site, siteBatchSize); if (pendingTasks.isEmpty()) { - return; // 没有待处理任务 + return; } - log.info("【打印队列】发现 {} 个待执行任务", pendingTasks.size()); + log.info("【打印队列】site={} 发现 {} 个待执行任务", site, pendingTasks.size()); - // 2. 按FIFO顺序逐个执行 for (PrintTask task : pendingTasks) { processTask(task); } - } catch (Exception e) { - log.error("打印任务调度器执行异常: {}", e.getMessage(), e); + log.error("【打印队列】site任务执行异常 site={}, error={}", site, e.getMessage(), e); + } finally { + siteProcessingFlag.set(false); } } @@ -75,6 +191,7 @@ public class PrintTaskScheduler { */ private void processTask(PrintTask task) { Long taskId = task.getId(); + String site = task.getSite(); String printerIp = task.getPrinterIp(); try { @@ -85,8 +202,8 @@ public class PrintTaskScheduler { return; } - log.info("【打印队列】开始执行任务 taskId={}, 打印机={}, 份数={}", - taskId, printerIp, task.getCopies()); + log.info("【打印队列】开始执行任务 taskId={}, site={}, 打印机={}, 份数={}", + taskId, site, printerIp, task.getCopies()); // 2. 解析标签数据 Map labelData = null; @@ -106,14 +223,14 @@ public class PrintTaskScheduler { // 4. 更新任务状态 if (success) { printTaskService.markAsSuccess(taskId); - log.info("【打印队列】✓ 任务完成 taskId={}, 打印机={}", taskId, printerIp); + log.info("【打印队列】✓ 任务完成 taskId={}, site={}, 打印机={}", taskId, site, printerIp); } else { printTaskService.markAsFailed(taskId, "打印失败，未返回成功标识"); - log.error("【打印队列】✗ 任务失败 taskId={}, 打印机={}", taskId, printerIp); + log.error("【打印队列】✗ 任务失败 taskId={}, site={}, 打印机={}", taskId, site, printerIp); } } catch (Exception e) { - log.error("【打印队列】✗ 任务执行异常 taskId={}, 错误: {}", taskId, e.getMessage(), e); + log.error("【打印队列】✗ 任务执行异常 taskId={}, site={}, 错误: {}", taskId, site, e.getMessage(), e); printTaskService.markAsFailed(taskId, "执行异常: " + e.getMessage()); } } @@ -134,8 +251,9 @@ public class PrintTaskScheduler { Socket socket = null; try { // 1. 连接打印机 - socket = new Socket(printerIp, 9100); - socket.setSoTimeout(10000); // 10秒超时 + socket = new Socket(); + socket.connect(new InetSocketAddress(printerIp, 9100), printerConnectTimeoutMs); + socket.setSoTimeout(printerReadTimeoutMs); OutputStream os = socket.getOutputStream(); @@ -145,7 +263,7 @@ public class PrintTaskScheduler { log.debug("开始发送ZPL到打印机 {}, 份数: {}, RFID: {}", printerIp, actualCopies, rfidFlag); for (int i = 0; i < actualCopies; i++) { - os.write(zplCode.getBytes("UTF-8")); + os.write(zplCode.getBytes(StandardCharsets.UTF_8)); os.flush(); // RFID标签需要等待处理时间