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

TASK:generate 修复逻辑Bug

Browse Source
This commit is contained in:
xupei
2024-01-25 13:37:17 +08:00
parent 65bed9dada
commit c35a188abb
7 changed files with 75 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java
View File

@@ -16,7 +16,7 @@ import org.springframework.beans.factory.annotation.Value;
public class MQConfig {
public static final String GENERATE_EXCHANGE_FANOUT = "generate-exchange";
public static final String GENERATE_QUEUE = "generate-queue";
public static final String GENERATE_QUEUE = "generate-queue-prod";
public MQConfig() {
}

14
src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java
View File

@@ -63,12 +63,12 @@ public class MQConsumer {
// 2.2 将该消息从取消列表中删除
// redisUtil.removeFromSet(cancelSetKey, uniqueId);
} else {
/*try {
Thread.sleep(15000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}*/
GenerateCollectionVO generateCollectionVO = generateService.generateThroughImageText(generateThroughImageTextDTO);
// try {
// Thread.sleep(15000);
// } catch (InterruptedException e) {
// throw new RuntimeException(e);
// }
// 将消息从redis排队队列中删除,需保证被消费的消息存储到db之后再从redis删除
redisUtil.removeFromZSet(consumptionOrderKey, uniqueId);
if (!Objects.isNull(generateCollectionVO)) {
@@ -100,8 +100,8 @@ public class MQConsumer {
}
long end = System.currentTimeMillis();
log.info(" task_id： " + uniqueId + "----------" + consumerName + " 执行时长：" + (end - start) + "毫秒");
log.info("============end listening==========");
log.info(" task_id： " + uniqueId + "----------" + consumerName + " 执行时长：" + (end - start) + "毫秒");
log.info("=============end listening===========");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)

15
src/main/java/com/ai/da/common/utils/AsyncCallerUtil.java
View File

@@ -30,21 +30,20 @@ public class AsyncCallerUtil {
public List<String> generate(GenerateToPythonDTO generateToPython) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
String taskId = generateToPython.getTasks_id();
waitingStatus.put(taskId, true);
ScheduledFuture<?> timeoutTask = null;
if (!waitingStatus.containsKey(taskId)) waitingStatus.put(taskId, true);
try {
CompletableFuture<List<String>> generateResult = callGenerateAsync(generateToPython);
// 10秒后第一次确认，之后每隔10秒确认一次用户选择结果
// 5秒后第一次确认，之后每隔10秒确认一次用户选择结果
timeoutTask = scheduledExecutorService.scheduleAtFixedRate(() -> {
// 调用另一个接口获取用户的选择
if (!waitingStatus.get(taskId)) {
// 如果用户选择取消，则取消对generate的调用，cancel判断是否成功取消
// 如果用户选择取消，则取消对generate的调用
generateResult.cancel(true);
waitingStatus.remove(taskId);
}
log.info("===============持续等待===============");
}, 10, 10, TimeUnit.SECONDS);
} else log.info("===============持续等待===============");
}, 5, 10, TimeUnit.SECONDS);
log.info("阻塞等待结果...");
// 阻塞，等待结果
@@ -53,6 +52,10 @@ public class AsyncCallerUtil {
timeoutTask.cancel(true);
waitingStatus.remove(taskId);
return result;
} catch (CancellationException e) {
// generateResult.cancel(true);通过抛出异常取消该任务
log.info("==========成功取消generate任务==========");
return null;
} catch (InterruptedException | ExecutionException | BusinessException e) {
// 处理异常
log.error("发生错误 ： " + e);