# 更新

data-common/src/main/java/com/bailuntec/mapper/DcBaseQueueMapper.java

@@ -162,4 +162,7 @@ public interface DcBaseQueueMapper {
      * @project https://github.com/itfsw/mybatis-generator-plugin
      */
     int upsertWithBLOBs(DcBaseQueue record);
-}
+
\ No newline at end of file
+
+    List<DcBaseQueue> poll(int shardingItem);
+}

data-show/show-auto-turnover/src/main/java/com/bailuntec/job/QueueConsumerJob.java

@@ -5,7 +5,6 @@ import com.alibaba.fastjson.annotation.JSONField;
 import com.bailuntec.domain.constant.CommonConstant;
 import com.bailuntec.domain.entity.DcBaseQueue;
 import com.bailuntec.domain.entity.DcBaseStock;
-import com.bailuntec.domain.example.DcBaseQueueExample;
 import com.bailuntec.domain.example.DcBaseStockExample;
 import com.bailuntec.mapper.DcBaseQueueMapper;
 import com.bailuntec.mapper.DcBaseStockMapper;
@@ -45,41 +44,36 @@ public class QueueConsumerJob implements SimpleJob {
     public void execute(ShardingContext shardingContext) {
 
         try (SqlSession sqlSession = SessionUtil.getFactory().openSession(true)) {
-            if (shardingContext.getShardingItem() == 0) {
-                AutoTurnoverJob autoTurnoverJob = new AutoTurnoverJob();
-                DcBaseQueueMapper dcBaseQueueMapper = sqlSession.getMapper(DcBaseQueueMapper.class);
-                DcBaseStockMapper dcBaseStockMapper = sqlSession.getMapper(DcBaseStockMapper.class);
+            AutoTurnoverJob autoTurnoverJob = new AutoTurnoverJob();
+            DcBaseQueueMapper dcBaseQueueMapper = sqlSession.getMapper(DcBaseQueueMapper.class);
+            DcBaseStockMapper dcBaseStockMapper = sqlSession.getMapper(DcBaseStockMapper.class);
 
-                List<DcBaseQueue> dcBaseQueueList = dcBaseQueueMapper.selectByExample(DcBaseQueueExample.newAndCreateCriteria()
-                        .andConsumeDateIsNull()
-                        .example()
-                        .limit(1000));
+            List<DcBaseQueue> dcBaseQueueList = dcBaseQueueMapper.poll(shardingContext.getShardingItem());
 
-                dcBaseQueueList.forEach(dcBaseQueue -> {
-                    BaseQueueMessage baseQueueMessage = JSON.parseObject(dcBaseQueue.getMessage(), BaseQueueMessage.class);
+            dcBaseQueueList.forEach(dcBaseQueue -> {
+                BaseQueueMessage baseQueueMessage = JSON.parseObject(dcBaseQueue.getMessage(), BaseQueueMessage.class);
 
-                    DcBaseStock dcBaseStock = dcBaseStockMapper.selectOneByExample(DcBaseStockExample.newAndCreateCriteria()
-                            .andBailunSkuEqualTo(baseQueueMessage.getBailunSku())
-                            .andWarehouseCodeEqualTo(baseQueueMessage.getWarehouseCode())
-                            .example());
+                DcBaseStock dcBaseStock = dcBaseStockMapper.selectOneByExample(DcBaseStockExample.newAndCreateCriteria()
+                        .andBailunSkuEqualTo(baseQueueMessage.getBailunSku())
+                        .andWarehouseCodeEqualTo(baseQueueMessage.getWarehouseCode())
+                        .example());
+                try {
+                    autoTurnoverJob.autoTurnoverFromStock(DateTimeFormatter.ofPattern(CommonConstant.DATE_FORMAT).format(LocalDate.now()), dcBaseStock);
+                    dcBaseQueueMapper.deleteByPrimaryKey(dcBaseQueue.getId());
+                } catch (Exception e) {
                     try {
-                        autoTurnoverJob.autoTurnoverFromStock(DateTimeFormatter.ofPattern(CommonConstant.DATE_FORMAT).format(LocalDate.now()), dcBaseStock);
-                        dcBaseQueueMapper.deleteByPrimaryKey(dcBaseQueue.getId());
-                    } catch (Exception e) {
-                        try {
-                            dcBaseQueue.setQueueType(2);
-                            dcBaseQueue.setErrorMessage(e.getMessage());
-                            dcBaseQueue.setErrorStackTrace(Arrays.toString(e.getStackTrace()));
-                            dcBaseQueueMapper.updateByPrimaryKey(dcBaseQueue);
-                        } catch (Exception ex) {
-                            log.error(ex.getMessage());
-                        }
-                    } finally {
-                        log.info("消费一条数据 message:{}", dcBaseQueue.getMessage());
+                        dcBaseQueue.setQueueType(2);
+                        dcBaseQueue.setErrorMessage(e.getMessage());
+                        dcBaseQueue.setErrorStackTrace(Arrays.toString(e.getStackTrace()));
+                        dcBaseQueueMapper.updateByPrimaryKey(dcBaseQueue);
+                    } catch (Exception ex) {
+                        log.error(ex.getMessage());
                     }
+                } finally {
+                    log.info("消费一条数据 message:{}", dcBaseQueue.getMessage());
+                }
 
-                });
-            }
+            });
         }
     }
 }