解决多线程并发

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/MqMessageRecordServiceImpl.java

@@ -133,18 +133,13 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 
             //根据aiAnalysisRequestId更新请求日志表的difyRequest字段
             syncUpdateDiFyRequest(difyCommunityTargetDTO, aiAnalysisRequestId);
-            
-            DiFyReq diFyReq = new DiFyReq();
-            diFyReq.setUser(user);
-            diFyReq.setInputs(difyCommunityTargetDTO);
             //调用DiFy工作流，并推送到社区
-            processDify(diFyReq, difyResult, aiAnalysisRequestId);
+            processDify(difyCommunityTargetDTO,user,difyResult, aiAnalysisRequestId);
         } catch (Exception e) {
             log.error("舆情自动化异常:{}", e.getMessage());
             //保存错误日志
             aiAnalysisErrorsMapper.insert(AiAnalysisErrors.builder()
                     .aiAnalysisRequestId(aiAnalysisRequestId)
-                    .difyResponse(difyResult.toJSONString())
                     .aiAnalysisErrorMessage(e.getMessage())
                     .aiAnalysisRequestType(BusinessTypeEnum.COMMUNITYTARGET.getCode())
                     .build());
@@ -152,33 +147,15 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
         return true;
     }
 
-    private void processDify(DiFyReq diFyReq, JSONArray difyResult, String aiAnalysisRequestId) throws InterruptedException, ExecutionException {
+    private void processDify(DifyCommunityTargetDTO difyCommunityTargetDTO ,String user, JSONArray difyResult, String aiAnalysisRequestId) throws InterruptedException, ExecutionException {
         //舆情案件分析
-        CompletableFuture<JSONObject> caseWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, caseToken),getAsyncExecutor)
+        CompletableFuture<JSONObject> caseWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, caseToken),getAsyncExecutor);
-                .exceptionally(ex -> {
-                    log.error("舆情案件分析失败: {}", ex.getMessage());
-                    JSONObject errorResult = new JSONObject();
-                    errorResult.put("error", "舆情案件分析失败: " + ex.getMessage());
-                    return errorResult;
-                });
 
         // 内容主题关键词打标
-        CompletableFuture<JSONObject> keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, keywordToken),getAsyncExecutor)
+        CompletableFuture<JSONObject> keywordWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, keywordToken),getAsyncExecutor);
-                .exceptionally(ex -> {
-                    log.error("内容主题关键词打标失败: {}", ex.getMessage());
-                    JSONObject errorResult = new JSONObject();
-                    errorResult.put("error", "内容主题关键词打标失败: " + ex.getMessage());
-                    return errorResult;
-                });
 
         // litecrm线索分析
-        CompletableFuture<JSONObject> clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(diFyReq, clueAnalysisToken),getAsyncExecutor)
+        CompletableFuture<JSONObject> clueAnalysisWorkFlow = CompletableFuture.supplyAsync(() -> callCommunityWorkFlow(difyCommunityTargetDTO,user, clueAnalysisToken),getAsyncExecutor);
-                .exceptionally(ex -> {
-                    log.error("litecrm线索分析失败: {}", ex.getMessage());
-                    JSONObject errorResult = new JSONObject();
-                    errorResult.put("error", "litecrm线索分析失败: " + ex.getMessage());
-                    return errorResult;
-                });
 
         CompletableFuture<Void> allFutures = CompletableFuture.allOf(caseWorkFlow, keywordWorkFlow, clueAnalysisWorkFlow);
         // 等待所有API调用完成
@@ -196,11 +173,12 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 
     /**
      * 调用案件，关键词，线索工作流
-     * @param diFyReq
-     * @param token
-     * @return
      */
-    private JSONObject callCommunityWorkFlow(DiFyReq diFyReq, String token) {
+    private JSONObject callCommunityWorkFlow(DifyCommunityTargetDTO difyCommunityTargetDTO ,String user, String token) {
+        log.info("开始调用案件，关键词，线索工作流,token: {}",token);
+        DiFyReq diFyReq = new DiFyReq();
+        diFyReq.setUser(user);
+        diFyReq.setInputs(difyCommunityTargetDTO);
         diFyReq.setFlowId(token);
         JSONObject difResult = (JSONObject) diFyService.getDiFyObject(diFyReq);
         return difResult;
@@ -263,6 +241,7 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
 
             caseResult.put("keyWords", keywordResult.getString("keyWords"));
             caseResult.put("targetTag", clueAnalysisResult.getString("targetTag"));
+            caseResult.put("aiAnalysisRequestId", aiAnalysisRequestId);
             //返回结果推送到社区的MQ
             rocketMQTemplate.syncSend(topic, caseResult.toString());
             log.info("舆情分析发送回调MQ完成: {}", caseResult);
@@ -650,11 +629,8 @@ public class MqMessageRecordServiceImpl extends ServiceImpl<MqMessageRecordMappe
                                 //根据aiAnalysisRequestId更新请求日志表的difyRequest字段
                                 syncUpdateDiFyRequest(difyCommunityTargetDTO, aiAnalysisError.getAiAnalysisRequestId());
 
-                                DiFyReq diFyReq = new DiFyReq();
-                                diFyReq.setUser(user);
-                                diFyReq.setInputs(difyCommunityTargetDTO);
                                 JSONArray difyResult = new JSONArray();
-                                processDify(diFyReq,difyResult,aiAnalysisError.getAiAnalysisRequestId());
+                                processDify(difyCommunityTargetDTO,user,difyResult,aiAnalysisError.getAiAnalysisRequestId());
 
                                 //根据ai_analysis_request_id更新ai_analysis_errors表中的retry_count字段+1,更新status字段为1
                                 aiAnalysisErrorsMapper.update(new AiAnalysisErrors(), new LambdaUpdateWrapper<AiAnalysisErrors>()