按站点（site）拆分任务，不同站点并发处理

src/main/java/com/gaotao/modules/base/dao/PrintTaskDao.java

@@ -25,6 +25,34 @@ public interface PrintTaskDao extends BaseMapper<PrintTask> {
             "ORDER BY created_date ASC")
     List<PrintTask> 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<String> 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<PrintTask> selectPendingTasksBySiteWithReadPast(@Param("site") String site,
+                                                          @Param("limit") int limit);
+
     /**
      * 将已完成的打印任务插入到历史表
      * 

src/main/java/com/gaotao/modules/base/service/Impl/PrintTaskServiceImpl.java

@@ -65,6 +65,18 @@ public class PrintTaskServiceImpl extends ServiceImpl<PrintTaskDao, PrintTask> i
         return this.baseMapper.selectPendingTasksWithReadPast(limit);
     }
 
+    @Override
+    public List<String> getPendingSites(int limit) {
+        // 查询有待执行任务的站点，按最早任务时间排序
+        return this.baseMapper.selectPendingSitesWithReadPast(limit);
+    }
+
+    @Override
+    public List<PrintTask> getPendingTasksBySite(String site, int limit) {
+        // 按站点查询待执行任务，确保站内FIFO顺序
+        return this.baseMapper.selectPendingTasksBySiteWithReadPast(site, limit);
+    }
+
     @Override
     @Transactional
     public boolean markAsProcessing(Long taskId) {

src/main/java/com/gaotao/modules/base/service/PrintTaskService.java

@@ -35,6 +35,23 @@ public interface PrintTaskService extends IService<PrintTask> {
      */
     List<PrintTask> getPendingTasks(int limit);
 
+    /**
+     * 获取有待执行任务的站点列表（按最早任务时间排序）
+     *
+     * @param limit 获取站点数量
+     * @return 待执行站点列表
+     */
+    List<String> getPendingSites(int limit);
+
+    /**
+     * 获取指定站点待执行的打印任务（FIFO顺序）
+     *
+     * @param site 站点编码
+     * @param limit 获取任务数量
+     * @return 待执行任务列表
+     */
+    List<PrintTask> getPendingTasksBySite(String site, int limit);
+
     /**
      * 更新任务为执行中
      * 

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;
 
 /**
  * 打印任务定时调度器
  *
  * <p><b>功能说明：</b></p>
  * <ul>
- *   <li>每秒扫描一次打印任务队列</li>
- *   <li>按FIFO顺序（created_date）取出PENDING状态的任务</li>
+ *   <li>每2秒扫描一次打印任务队列</li>
+ *   <li>按站点（site）拆分任务，不同站点并发处理</li>
+ *   <li>同一站点内按FIFO顺序（created_date）取出PENDING状态的任务</li>
  *   <li>执行打印任务并更新状态</li>
  *   <li>失败任务记录错误信息</li>
  * </ul>
@@ -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<String, AtomicBoolean> 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<PrintTask> pendingTasks = printTaskService.getPendingTasks(10);
+            // 1. 获取有待执行任务的站点列表
+            List<String> 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<PrintTask> 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<String, Object> 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标签需要等待处理时间