修改邮件定时任务

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/config/AsyncConfig.java

@@ -1,10 +1,40 @@
 package com.storlead.trade.config;
 
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
 
 @Configuration
 @EnableAsync
 public class AsyncConfig {
     // 启用 Spring 异步支持
-}
+
+    /**
+     * 邮件内容生成（AI 重写正文）专用线程池。
+     * <p>
+     * 之前没有自定义 executor，Spring 会退化为 SimpleAsyncTaskExecutor：
+     * 每个任务 new 一条线程，一个 SOP 步骤几十上百封邮件就会瞬间起几十上百条线程打 AI，
+     * 既容易把 AI 服务打爆，也会拖垮本机。这里限制并发，并用队列兜底。
+     * <p>
+     * 队列满时走 CallerRunsPolicy：由提交任务的线程（请求线程）自己执行，任务不会丢，
+     * 但会退化成同步、接口变慢——日志里能看到线程名变成 http-nio-xxx 而不是 trade-email-x。
+     */
+    @Bean("tradeEmailExecutor")
+    public Executor tradeEmailExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(4);
+        executor.setMaxPoolSize(8);
+        executor.setQueueCapacity(2000);
+        executor.setKeepAliveSeconds(60);
+        executor.setThreadNamePrefix("trade-email-");
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        executor.setAwaitTerminationSeconds(30);
+        executor.initialize();
+        return executor;
+    }
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/service/MarketEmailsService.java

@@ -160,6 +160,12 @@ public interface MarketEmailsService extends MyBaseService<MarketEmailsEntity> {
      */
     int cancelOverduePending();
 
+    /**
+     * 异步生成邮件正文并落库（一次 AI 调用）。
+     * <p>
+     * 真正生效的 @Async 在实现类方法上（{@code MarketEmailsServiceImpl#updateEmailContent}），
+     * 并指定了 {@code tradeEmailExecutor} 线程池；这里的注解仅作说明，保留以免漏看。
+     */
     @Async
     void updateEmailContent(MarketEmailsEntity marketEmailsEntity);
 }

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/service/impl/MarketEmailsServiceImpl.java

@@ -33,6 +33,7 @@ import org.slf4j.Logger;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
@@ -538,42 +539,62 @@ public class MarketEmailsServiceImpl extends MyBaseServiceImpl<MarketEmailsMappe
         return rows;
     }
 
+    /**
+     * 生成邮件正文（调 AI 重写内容）并落库。
+     * <p>
+     * 一次调用 = 一次 AI 请求，耗时通常是秒级，一个 SOP 步骤下可能有几十上百封邮件，
+     * 所以这里必须是异步的：@Async 放在<strong>实现类方法</strong>上（放在接口方法上时，
+     * 一旦 bean 走 CGLIB 代理或与其它 AOP 代理叠加，匹配行为会随代理方式变化，不如放实现类稳妥）。
+     * <p>
+     * 异步线程的异常不会回到请求线程，因此内部自行兜住并记 error 日志，
+     * 避免失败后邮件一直停在旧内容上却没有任何痕迹。
+     */
     @Override
+    @Async("tradeEmailExecutor")
     public void updateEmailContent(MarketEmailsEntity marketEmailsEntity) {
-        //通过客户id获取客户分析结果
-        CustomerAnalysisResultEntity customerAnalysisResult = new CustomerAnalysisResultEntity();
-        LambdaQueryWrapper<CustomerAnalysisResultEntity> queryWrapper = new LambdaQueryWrapper<>();
-        queryWrapper.eq(CustomerAnalysisResultEntity::getCustomerId, marketEmailsEntity.getCustomerId());
-        queryWrapper.last("limit 1");
-        customerAnalysisResult = customerAnalysisResultService.getOne(queryWrapper);
-        String reasonSummary = "";
-        if (customerAnalysisResult != null){
-            reasonSummary = customerAnalysisResult.getReasonSummary();
-        }
-        Map sendMap = new HashMap();
-        sendMap.put("email", marketEmailsEntity.getContent());
-        sendMap.put("reason_summary", reasonSummary);
-
-        ChatDTO chatDTO = new ChatDTO();
-        chatDTO.setInputs(sendMap);
-//            chatDTO.setQuery("send_time");
-        chatDTO.setResponseMode("blocking");
-        chatDTO.setAppId("app-CVPtLUPO0zd2SeGPvZv3Q4Ev");
-        chatDTO.setUser("1");
-
-        // 调取 AI  获取新的content
-        Result<Object> r = workflowsService.AiWorkflows(chatDTO);
-        if (r == null) {
-            log.warn("无内容生成");
-            return;
-        }
-        // 从结构化结果里取出 AI 真正输出的 text 字段（多个结束节点拼接）。
-        // 不再做 inner.toString() 兜底——整串 Result 里有 files=[]/inputs=[] 等空数组
-        // 会被误当成 AI 输出的 JSON 数组（如 [models]），污染解析。
-        String text = extractCleanText(r.getResult());
-        marketEmailsEntity.setContent(text);
+        log.info("[updateEmailContent] 提交异步生成，emailId={}，customerId={}，线程={}",
+                marketEmailsEntity.getId(), marketEmailsEntity.getCustomerId(), Thread.currentThread().getName());
+        try {
+            //通过客户id获取客户分析结果
+            CustomerAnalysisResultEntity customerAnalysisResult = new CustomerAnalysisResultEntity();
+            LambdaQueryWrapper<CustomerAnalysisResultEntity> queryWrapper = new LambdaQueryWrapper<>();
+            queryWrapper.eq(CustomerAnalysisResultEntity::getCustomerId, marketEmailsEntity.getCustomerId());
+            queryWrapper.last("limit 1");
+            customerAnalysisResult = customerAnalysisResultService.getOne(queryWrapper);
+            String reasonSummary = "";
+            if (customerAnalysisResult != null) {
+                reasonSummary = customerAnalysisResult.getReasonSummary();
+            }
+            Map sendMap = new HashMap();
+            sendMap.put("email", marketEmailsEntity.getContent());
+            sendMap.put("reason_summary", reasonSummary);
+
+            ChatDTO chatDTO = new ChatDTO();
+            chatDTO.setInputs(sendMap);
+            chatDTO.setResponseMode("blocking");
+            chatDTO.setAppId("app-CVPtLUPO0zd2SeGPvZv3Q4Ev");
+            chatDTO.setUser("1");
 
-        this.updateById(marketEmailsEntity);
+            // 调取 AI  获取新的content
+            Result<Object> r = workflowsService.AiWorkflows(chatDTO);
+            if (r == null) {
+                log.warn("无内容生成，emailId={}", marketEmailsEntity.getId());
+                return;
+            }
+            // 从结构化结果里取出 AI 真正输出的 text 字段（多个结束节点拼接）。
+            // 不再做 inner.toString() 兜底——整串 Result 里有 files=[]/inputs=[] 等空数组
+            // 会被误当成 AI 输出的 JSON 数组（如 [models]），污染解析。
+            String text = extractCleanText(r.getResult());
+            marketEmailsEntity.setContent(text);
+
+            this.updateById(marketEmailsEntity);
+            log.info("[updateEmailContent] 生成完成并已落库，emailId={}，线程={}",
+                    marketEmailsEntity.getId(), Thread.currentThread().getName());
+        } catch (Exception e) {
+            // 异步线程：异常不会回到请求线程，这里必须自己兜住，否则失败将毫无痕迹
+            log.error("[updateEmailContent] 生成邮件内容失败，emailId={}，customerId={}",
+                    marketEmailsEntity.getId(), marketEmailsEntity.getCustomerId(), e);
+        }
     }
 
     /**