webcad异步导入恢复谁发消息谁消费的本地缓存模式

cf-framework/cf-spring-boot-starter-mq/src/main/java/com/cf/imes/framework/mq/rabbitmq/constant/RabbitMqConstants.java

@@ -36,7 +36,7 @@ public class RabbitMqConstants {
     /**
      * webcad拆单队列
      */
-    public static final String ORDER_IMPORT_WEBCAD_QUEUE = "order_import_webcad_queue";
+    public static final String ORDER_IMPORT_WEBCAD_QUEUE = "order_import_webcad_queue_%s";
 
     /**
      * 生产单导入死信交换机

cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/config/ChenfengRabbitMQPlanConfiguration.java

@@ -7,6 +7,7 @@ import org.springframework.amqp.core.DirectExchange;
 import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -16,6 +17,7 @@ import java.util.Map;
 import static com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants.ORDER_IMPORT_DEAD_LETTER_EXCHANGE;
 import static com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants.ORDER_IMPORT_DEAD_LETTER_QUEUE;
 import static com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants.ORDER_IMPORT_DEAD_LETTER_ROUTING_KEY;
+import static com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants.ORDER_IMPORT_WEBCAD_QUEUE;
 
 /**
  * @author Gqr
@@ -23,18 +25,20 @@ import static com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants.ORDER
  */
 @Configuration
 public class ChenfengRabbitMQPlanConfiguration {
+    @Value("${chenfeng.plan.webcad.import.node:1}")
+    private String instanceId;
 
     private static final String X_DEAD_LETTER_EXCHANGE_KEY = "x-dead-letter-exchange";
     private static final String X_DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
 
-    /**
-     * webcad拆单导入交换机、队列、绑定声明
-     *
-     * @return
-     */
     @Bean
-    public FanoutExchange orderImportWebCadFanoutExchange() {
+    public DirectExchange orderImportWebCadFanoutExchange() {
-        return new FanoutExchange(RabbitMqConstants.ORDER_IMPORT_WEBCAD_EXCHANGE);
+        return new DirectExchange(RabbitMqConstants.ORDER_IMPORT_WEBCAD_EXCHANGE);
+    }
+
+    @Bean("orderImportWebCadQueueName")
+    public String dynamicOrderImportWebCadQueueName() {
+        return String.format(ORDER_IMPORT_WEBCAD_QUEUE, instanceId);
     }
 
     /**
@@ -49,7 +53,7 @@ public class ChenfengRabbitMQPlanConfiguration {
         args.put(X_DEAD_LETTER_EXCHANGE_KEY, ORDER_IMPORT_DEAD_LETTER_EXCHANGE);
         // 设置死信路由键
         args.put(X_DEAD_LETTER_ROUTING_KEY, ORDER_IMPORT_DEAD_LETTER_ROUTING_KEY);
-        return QueueBuilder.durable(RabbitMqConstants.ORDER_IMPORT_WEBCAD_QUEUE).withArguments(args).build();
+        return QueueBuilder.durable(dynamicOrderImportWebCadQueueName()).withArguments(args).build();
     }
 
     /**
@@ -59,7 +63,7 @@ public class ChenfengRabbitMQPlanConfiguration {
      */
     @Bean
     public Binding orderImportWebCadBinding() {
-        return BindingBuilder.bind(orderImportWebCadQueue()).to(orderImportWebCadFanoutExchange());
+        return BindingBuilder.bind(orderImportWebCadQueue()).to(orderImportWebCadFanoutExchange()).with(instanceId);
     }
 
     /**

cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/WebCadOrderImportServiceImpl.java

@@ -1,5 +1,7 @@
 package com.cf.imes.module.plan.service.orderImport;
 
+import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.io.IORuntimeException;
 import cn.hutool.core.util.NumberUtil;
 import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.nacos.common.utils.StringUtils;
@@ -55,15 +57,10 @@ import jakarta.annotation.Resource;
 
 import java.io.File;
 import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
 import java.util.zip.GZIPInputStream;
 
 import static com.cf.imes.module.executor.enums.ErrorCodeConstants.ORDER_IMPORT_ORGAN_LOCK_ERROR;
 import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_IMPORT_FAILED;
-import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_IMPORT_ORGANID_EMPTY_ERROR;
 import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_NOT_EXISTS_ERROR;
 
 /**
@@ -263,18 +260,11 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
                 .organId(organId)
                 .build();
         orderImportTaskMapper.insert(orderImportTaskDO);
-        // 缓存到nfs目录
+        // 缓存请求到本地磁盘
-        Path targetFile = Path.of(webCadImportProperties.getTempFilePath(), File.separator + taskId + ".json");
-        Path tmpFile = Path.of(webCadImportProperties.getTempFilePath(), File.separator + taskId + ".json.tmp");
         ObjectMapper objectMapper = new ObjectMapper();
+        File file = new File(webCadImportProperties.getTempFilePath() + File.separator + taskId + ".json");
+        objectMapper.writeValue(file, node);
 
-        // 原子写入：先写临时文件，再 move
-        try (OutputStream out = Files.newOutputStream(tmpFile)) {
-            objectMapper.writeValue(out, node);
-            out.flush();
-        }
-
-        Files.move(tmpFile, targetFile, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
         return taskId;
     }
 
@@ -295,12 +285,12 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
                 @Override
                 public void afterCommit() {
                     // 数据库操作全部完成发送消息
-                    rabbitTemplate.convertAndSend(RabbitMqConstants.ORDER_IMPORT_WEBCAD_EXCHANGE, null, message);
+                    rabbitTemplate.convertAndSend(RabbitMqConstants.ORDER_IMPORT_WEBCAD_EXCHANGE, webCadImportProperties.getNode(), message);
                 }
             });
         } else {
             // 数据库操作全部完成发送消息
-            rabbitTemplate.convertAndSend(RabbitMqConstants.ORDER_IMPORT_WEBCAD_EXCHANGE, null, message);
+            rabbitTemplate.convertAndSend(RabbitMqConstants.ORDER_IMPORT_WEBCAD_EXCHANGE, webCadImportProperties.getNode(), message);
         }
     }
 
@@ -334,12 +324,12 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
      */
     private void removeCacheFileWhenException(boolean sync, Long taskId) {
         if (!sync && ObjectUtil.isNotNull(taskId)) {
-            Path file = Path.of(webCadImportProperties.getTempFilePath(), taskId + ".json");
+            String filePath = webCadImportProperties.getTempFilePath() + File.separator + taskId + ".json";
             try {
-                Files.deleteIfExists(file); // 文件不存在也不会抛异常
+                FileUtil.del(filePath);
-            } catch (IOException e) {
+            } catch (IORuntimeException e) {
                 // 可以记录日志
-                log.warn("删除任务缓存文件失败: {}", file, e);
+                log.warn("删除任务缓存文件失败: {}", filePath, e);
             }
         }
     }

cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/consumer/WebCadOrderImportConsumer.java

@@ -107,7 +107,7 @@ public class WebCadOrderImportConsumer {
     @Resource
     private ChenfengCacheProperties chenfengCacheProperties;
 
-    @RabbitListener(queues = RabbitMqConstants.ORDER_IMPORT_WEBCAD_QUEUE)
+    @RabbitListener(queues = "#{@orderImportWebCadQueueName}")
     @DS("imes_prod")
     @Transactional(rollbackFor = Exception.class)
     public void orderImport(String message,

cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/property/WebCadImportProperties.java

@@ -23,6 +23,11 @@ public class WebCadImportProperties {
      */
     private int asyncPlateThreshold;
 
+    /**
+     * 当前拆单的节点，异步用
+     */
+    private String node;
+
     /**
      * 异步拆单请求json缓存地址
      */
@@ -36,6 +41,14 @@ public class WebCadImportProperties {
         this.tempFilePath = tempFilePath;
     }
 
+    public String getNode() {
+        return node;
+    }
+
+    public void setNode(String node) {
+        this.node = node;
+    }
+
     public int getAsyncPlateThreshold() {
         return asyncPlateThreshold;
     }

cf-module-prod-plan/cf-module-prod-plan-biz/src/main/resources/application-dev.yaml

@@ -166,9 +166,10 @@ chenfeng:
   plan:
     webcad:
       import:
+        node: 1
         async-plate-threshold: 5000
         cadImportPlateNumThreshold: 20000
-        temp-file-path: /data/imes-server/nfs
+        temp-file-path: /data/imes-server/cf-prod-plan/local
 
 mybatis-plus:
   configuration:

cf-module-prod-plan/cf-module-prod-plan-biz/src/main/resources/application-local.yaml

@@ -194,9 +194,10 @@ chenfeng:
   plan:
     webcad:
       import:
-        async-plate-threshold: 1000
+        node: 1
+        async-plate-threshold: 5000
         cadImportPlateNumThreshold: 20000
-        temp-file-path: Z:/
+        temp-file-path: D:\\local
 
 mybatis-plus:
   configuration: