aidlab/aida_back
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev/dev_xp' into dev/dev

Browse Source
This commit is contained in:
xupei
2024-04-23 16:46:55 +08:00
parent e7d4e98ad0 2882b06128
commit 1e09508efb
3 changed files with 32 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/main/java/com/ai/da/common/RabbitMQ/GenerateConsumer.java
View File

@@ -105,19 +105,20 @@ public class GenerateConsumer {
log.info("=============end listening===========");
}
public void processGenerateResult(Message msg, Channel channel){
public void processGenerateResult(Message msg, Channel channel) {
log.info("============ProcessGenerateResult listening==========");
long start = System.currentTimeMillis();
Map<String, String> generateResult = JSONObject.parseObject(msg.getBody(), Map.class);
log.info("generate response : {}", generateResult);
try{
log.info("tasks_id : {} start ",generateResult.get("tasks_id"));
if (generateResult.get("status").equals("SUCCESS")){
try {
log.info("tasks_id : {} start ", generateResult.get("tasks_id"));
if (generateResult.get("status").equals("SUCCESS")) {
String url = generateResult.get("data");
String taskId = generateResult.get("tasks_id");
generateService.processGenerateResult(taskId, url);
}else {
generateService.processGenerateResult(taskId, url);
} else {
// 修改redis中的数据状态为exception
String key = generateResultKey + ":" + generateResult.get("tasks_id");
redisUtil.addToString(key, new Gson().toJson(new GenerateResultVO(generateResult.get("tasks_id"), null, null, "Fail")), CommonConstant.GENERATE_RESULT_EXPIRE_TIME);
@@ -127,7 +128,7 @@ public class GenerateConsumer {
// 存redis
redisUtil.addToMap(exceptionMapKey, exceptionInfo);
}
}catch (Exception e){
} catch (Exception e) {
log.error(e.getMessage());
try {
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
@@ -146,7 +147,7 @@ public class GenerateConsumer {
}
long end = System.currentTimeMillis();
log.info("tasks_id : {}, message : {}, 执行时长： {} 毫秒",generateResult.get("tasks_id"), generateResult.get("message"), (end - start));
log.info("tasks_id : {}, end , message : {}, 执行时长： {} 毫秒", generateResult.get("tasks_id"), generateResult.get("message"), (end - start));
log.info("============ProcessGenerateResult End listening==========");
}
@@ -207,7 +208,7 @@ public class GenerateConsumer {
@RabbitListener(queues = MQConfig.GENERATE_RESULT_QUEUE)
@RabbitHandler
public void getGenerateResult(Message msg, Channel channel){
public void getGenerateResult(Message msg, Channel channel) {
processGenerateResult(msg, channel);
}
}

18
src/main/java/com/ai/da/common/RabbitMQ/SRConsumer.java
View File

@@ -132,17 +132,31 @@ public class SRConsumer {
try {
result = JSONObject.parseObject(msg.getBody(), JSONObject.class);
log.info("SR response : {}", result);
taskId = result.get("tasks_id").toString();
} catch (JSONException e) {
log.error("SRResult 返回数据格式不合规范");
log.error(e.getMessage());
setErrorMessage(msg, channel, e.getMessage(), null);
try {
// 第二个参数，是否批量确认消息，当传false时，只确认当前 deliveryTag对应的消息;当传true时，会确认当前及之前所有未确认的消息。
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
} catch (IOException exception) {
log.error("手动确认，取消返回队列，不再重新消费");
}
} catch (Exception e){
log.error(e.getMessage());
setErrorMessage(msg, channel, e.getMessage(), null);
try {
// 第二个参数，是否批量确认消息，当传false时，只确认当前 deliveryTag对应的消息;当传true时，会确认当前及之前所有未确认的消息。
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
} catch (IOException exception) {
log.error("手动确认，取消返回队列，不再重新消费");
}
}
try {
// channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
// 2、判断状态是否成功
if ("SUCCESS".equals(result.get("status").toString())) {
String output = result.get("data").toString();

6
src/main/java/com/ai/da/service/impl/GenerateServiceImpl.java
View File

@@ -38,6 +38,7 @@ import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
import static com.ai.da.common.enums.CollectionLevel1TypeEnum.*;
@@ -527,6 +528,11 @@ public class GenerateServiceImpl extends ServiceImpl<GenerateMapper, Generate> i
}
results.add(generateResultVO);
});
// todo
Set<String> collect = results.stream().map(GenerateResultVO::getStatus).collect(Collectors.toSet());
if (taskIdList.size() == 4 && collect.size() == 1 && collect.contains("Fail")){
throw new BusinessException("generate.interface.error");
}
return results;
}