diff --git a/backend/src/main/java/com/imeeting/common/RedisKeys.java b/backend/src/main/java/com/imeeting/common/RedisKeys.java index 9faf822..68b15ce 100644 --- a/backend/src/main/java/com/imeeting/common/RedisKeys.java +++ b/backend/src/main/java/com/imeeting/common/RedisKeys.java @@ -51,8 +51,12 @@ public final class RedisKeys { return "biz:meeting:asr:schedule:lock"; } - public static String meetingAsrPermitSetKey() { - return "biz:meeting:asr:permit:set"; + public static String meetingAsrPermitSetKey(String queueKey) { + return "biz:meeting:asr:permit:set:" + normalizeQueueKey(queueKey); + } + + public static String meetingAsrPermitQueueKey(Long meetingId) { + return "biz:meeting:asr:permit:meeting:" + meetingId; } public static String meetingAsrPermitSyncLockKey() { @@ -131,6 +135,13 @@ public final class RedisKeys { return "biz:android:device:weather:" + cityName; } + private static String normalizeQueueKey(String queueKey) { + if (queueKey == null || queueKey.isBlank()) { + return "unknown"; + } + return queueKey.trim(); + } + public static final String CACHE_EMPTY_MARKER = "EMPTY_MARKER"; public static final String SYS_PARAM_FIELD_VALUE = "value"; public static final String SYS_PARAM_FIELD_TYPE = "type"; diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java index cfe86f9..97b875d 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java @@ -282,7 +282,8 @@ public class AiTaskServiceImpl extends ServiceImpl impleme asrTask.setQueuedAt(LocalDateTime.now()); this.updateById(asrTask); } - if (!meetingAsrPermitCache.hasPermit(meetingId)) { + String asrQueueKey = resolveAsrQueueKey(asrTask); + if (!meetingAsrPermitCache.hasPermit(meetingId, asrQueueKey)) { log.info("[ASR-FLOW] ASR任务处于排队状态，等待调度执行: meetingId={}, asrTaskId={}", meetingId, asrTask.getId()); meetingProgressService.markQueued(meetingId, asrTask, 1, "ASR queued and waiting for execution"); return; @@ -484,10 +485,12 @@ public class AiTaskServiceImpl extends ServiceImpl impleme if (getBaseMapper() == null) { return; } - long queuedCount = count(new LambdaQueryWrapper() + List queuedTasks = list(new LambdaQueryWrapper() .eq(AiTask::getTaskType, "ASR") - .eq(AiTask::getStatus, 0)); - if (queuedCount <= 0) { + .eq(AiTask::getStatus, 0) + .orderByAsc(AiTask::getQueuedAt) + .orderByAsc(AiTask::getId)); + if (queuedTasks.isEmpty()) { return; } boolean acquired = meetingLockCache.tryAcquireAsrScheduleLock(Duration.ofSeconds(30)); @@ -496,28 +499,34 @@ public class AiTaskServiceImpl extends ServiceImpl impleme } try { int maxConcurrent = resolveAsrMaxConcurrent(); - long runningCount = Math.max(count(new LambdaQueryWrapper() + List runningTasks = list(new LambdaQueryWrapper() .eq(AiTask::getTaskType, "ASR") - .eq(AiTask::getStatus, 1)), meetingAsrPermitCache.countPermits()); - int available = (int) (maxConcurrent - runningCount); - if (available <= 0) { - return; + .eq(AiTask::getStatus, 1)); + Map asrModelCache = new HashMap<>(); + Map runningCountByQueue = buildAsrQueueCountMap(runningTasks, asrModelCache); + Map availableByQueue = new HashMap<>(); + for (AiTask queuedTask : queuedTasks) { + String queueKey = resolveAsrQueueKey(queuedTask, asrModelCache); + long occupied = Math.max( + runningCountByQueue.getOrDefault(queueKey, 0L), + meetingAsrPermitCache.countPermits(queueKey) + ); + availableByQueue.putIfAbsent(queueKey, Math.max(0, maxConcurrent - (int) occupied)); } - List queuedTasks = list(new LambdaQueryWrapper() - .eq(AiTask::getTaskType, "ASR") - .eq(AiTask::getStatus, 0) - .orderByAsc(AiTask::getQueuedAt) - .orderByAsc(AiTask::getId) - .last("LIMIT " + available)); List claimedTasks = new ArrayList<>(); for (AiTask queuedTask : queuedTasks) { + String queueKey = resolveAsrQueueKey(queuedTask, asrModelCache); + if (availableByQueue.getOrDefault(queueKey, 0) <= 0) { + continue; + } // 当前会议仍持有轮询锁时，不能提前 claim 队列任务。 // 否则任务会先变成 RUNNING，再被异步 dispatch 因拿不到同一把锁而直接跳过，后续也不会再回到队列。 if (meetingLockCache.hasPollingLock(queuedTask.getMeetingId())) { continue; } - if (claimQueuedAsrTaskForScheduling(queuedTask)) { + if (claimQueuedAsrTaskForScheduling(queuedTask, queueKey)) { claimedTasks.add(queuedTask); + availableByQueue.put(queueKey, availableByQueue.get(queueKey) - 1); } } refreshQueuedAsrProgress(); @@ -548,11 +557,11 @@ public class AiTaskServiceImpl extends ServiceImpl impleme } } - private boolean claimQueuedAsrTaskForScheduling(AiTask task) { + private boolean claimQueuedAsrTaskForScheduling(AiTask task, String queueKey) { if (task == null || task.getMeetingId() == null || task.getId() == null || !Integer.valueOf(0).equals(task.getStatus())) { return false; } - return meetingAsrPermitCache.acquirePermit(task.getMeetingId()); + return meetingAsrPermitCache.acquirePermit(task.getMeetingId(), queueKey); } private int resolveAsrMaxConcurrent() { @@ -623,6 +632,70 @@ public class AiTaskServiceImpl extends ServiceImpl impleme } } + private Map buildAsrQueueCountMap(List tasks, Map asrModelCache) { + Map counts = new HashMap<>(); + if (tasks == null || tasks.isEmpty()) { + return counts; + } + for (AiTask task : tasks) { + String queueKey = resolveAsrQueueKey(task, asrModelCache); + counts.merge(queueKey, 1L, Long::sum); + } + return counts; + } + + private String resolveAsrQueueKey(AiTask task) { + return resolveAsrQueueKey(task, new HashMap<>()); + } + + private String resolveAsrQueueKey(AiTask task, Map asrModelCache) { + Long asrModelId = extractAsrModelId(task); + Long meetingId = task == null ? null : task.getMeetingId(); + if (asrModelId == null) { + return buildFallbackAsrQueueKey(null, meetingId); + } + AiModelVO asrModel = resolveAsrModel(asrModelId, asrModelCache); + if (asrModel != null) { + String provider = firstNonBlank(asrModel.getProvider()); + if (TENCENT_PROVIDER.equalsIgnoreCase(provider)) { + return "tencent://offline-asr/" + TENCENT_ASR_REGION; + } + String baseUrl = firstNonBlank(asrModel.getBaseUrl()); + if (baseUrl != null) { + return appendPath(baseUrl, "api/v1/asr/transcriptions"); + } + } + return buildFallbackAsrQueueKey(asrModelId, meetingId); + } + + private AiModelVO resolveAsrModel(Long asrModelId, Map asrModelCache) { + if (asrModelId == null) { + return null; + } + if (asrModelCache == null) { + return aiModelService.getModelById(asrModelId, "ASR"); + } + AiModelVO cached = asrModelCache.get(asrModelId); + if (cached != null) { + return cached; + } + AiModelVO model = aiModelService.getModelById(asrModelId, "ASR"); + if (model != null) { + asrModelCache.put(asrModelId, model); + } + return model; + } + + private String buildFallbackAsrQueueKey(Long asrModelId, Long meetingId) { + if (asrModelId != null) { + return "asr-model://" + asrModelId; + } + if (meetingId != null) { + return "meeting://" + meetingId; + } + return "asr://unknown"; + } + private String extractExternalTaskId(AiTask task) { if (task == null || task.getResponseData() == null || task.getResponseData().get("task_id") == null) { return null; diff --git a/backend/src/main/java/com/imeeting/support/redis/MeetingAsrPermitCache.java b/backend/src/main/java/com/imeeting/support/redis/MeetingAsrPermitCache.java index aae4142..88fb8ce 100644 --- a/backend/src/main/java/com/imeeting/support/redis/MeetingAsrPermitCache.java +++ b/backend/src/main/java/com/imeeting/support/redis/MeetingAsrPermitCache.java @@ -3,11 +3,13 @@ package com.imeeting.support.redis; import com.imeeting.common.RedisKeys; import com.imeeting.support.RedisSupport; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.util.List; @Component +@Slf4j @RequiredArgsConstructor public class MeetingAsrPermitCache { @@ -18,33 +20,60 @@ public class MeetingAsrPermitCache { RedisKeys.meetingAsrPermitSyncLockKey(), RedisKeys.meetingAsrRefillLockKey() )); - if (meetingId != null) { - redisSupport.removeFromSetQuietly(RedisKeys.meetingAsrPermitSetKey(), String.valueOf(meetingId)); - } + removePermit(meetingId); } public void removePermit(Long meetingId) { if (meetingId == null) { return; } - redisSupport.removeFromSetQuietly(RedisKeys.meetingAsrPermitSetKey(), String.valueOf(meetingId)); + String queueKey = redisSupport.getStringQuietly(RedisKeys.meetingAsrPermitQueueKey(meetingId)); + if (queueKey != null && !queueKey.isBlank()) { + redisSupport.removeFromSetQuietly(RedisKeys.meetingAsrPermitSetKey(queueKey), String.valueOf(meetingId)); + } + redisSupport.deleteQuietly(RedisKeys.meetingAsrPermitQueueKey(meetingId)); } - public boolean acquirePermit(Long meetingId) { - if (meetingId == null) { + public boolean acquirePermit(Long meetingId, String queueKey) { + String normalizedQueueKey = normalizeQueueKey(queueKey); + if (meetingId == null || normalizedQueueKey == null) { return false; } - return redisSupport.addToSetQuietly(RedisKeys.meetingAsrPermitSetKey(), String.valueOf(meetingId)); - } - - public boolean hasPermit(Long meetingId) { - if (meetingId == null) { + String permitSetKey = RedisKeys.meetingAsrPermitSetKey(normalizedQueueKey); + String member = String.valueOf(meetingId); + if (!redisSupport.addToSetQuietly(permitSetKey, member)) { + return false; + } + try { + redisSupport.setString(RedisKeys.meetingAsrPermitQueueKey(meetingId), normalizedQueueKey); + return true; + } catch (RuntimeException ex) { + redisSupport.removeFromSetQuietly(permitSetKey, member); + log.warn("record ASR permit queue key failed, meetingId={}, queueKey={}", meetingId, normalizedQueueKey, ex); return false; } - return redisSupport.isSetMemberQuietly(RedisKeys.meetingAsrPermitSetKey(), String.valueOf(meetingId)); } - public long countPermits() { - return redisSupport.getSetSizeQuietly(RedisKeys.meetingAsrPermitSetKey()); + public boolean hasPermit(Long meetingId, String queueKey) { + String normalizedQueueKey = normalizeQueueKey(queueKey); + if (meetingId == null || normalizedQueueKey == null) { + return false; + } + return redisSupport.isSetMemberQuietly(RedisKeys.meetingAsrPermitSetKey(normalizedQueueKey), String.valueOf(meetingId)); + } + + public long countPermits(String queueKey) { + String normalizedQueueKey = normalizeQueueKey(queueKey); + if (normalizedQueueKey == null) { + return 0L; + } + return redisSupport.getSetSizeQuietly(RedisKeys.meetingAsrPermitSetKey(normalizedQueueKey)); + } + + private String normalizeQueueKey(String queueKey) { + if (queueKey == null || queueKey.isBlank()) { + return null; + } + return queueKey.trim(); } }