chenh
/
imeeting
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix:ASR排队优化 全租户排队改为按模型前缀URL进行排队

dev_na
chenhao 2026-07-10 10:01:08 +08:00
parent 9e12e03481
commit 0aaa3a1cde
3 changed files with 147 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
backend/src/main/java/com/imeeting/common/RedisKeys.java
View File

@ -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";

109
backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java
View File

@ -282,7 +282,8 @@ public class AiTaskServiceImpl extends ServiceImpl<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> impleme
if (getBaseMapper() == null) {
return;
}
long queuedCount = count(new LambdaQueryWrapper<AiTask>()
List<AiTask> queuedTasks = list(new LambdaQueryWrapper<AiTask>()
.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<AiTaskMapper, AiTask> impleme
}
try {
int maxConcurrent = resolveAsrMaxConcurrent();
long runningCount = Math.max(count(new LambdaQueryWrapper<AiTask>()
List<AiTask> runningTasks = list(new LambdaQueryWrapper<AiTask>()
.eq(AiTask::getTaskType, "ASR")
.eq(AiTask::getStatus, 1)), meetingAsrPermitCache.countPermits());
int available = (int) (maxConcurrent - runningCount);
if (available <= 0) {
return;
.eq(AiTask::getStatus, 1));
Map<Long, AiModelVO> asrModelCache = new HashMap<>();
Map<String, Long> runningCountByQueue = buildAsrQueueCountMap(runningTasks, asrModelCache);
Map<String, Integer> 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<AiTask> queuedTasks = list(new LambdaQueryWrapper<AiTask>()
.eq(AiTask::getTaskType, "ASR")
.eq(AiTask::getStatus, 0)
.orderByAsc(AiTask::getQueuedAt)
.orderByAsc(AiTask::getId)
.last("LIMIT " + available));
List<AiTask> 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<AiTaskMapper, AiTask> 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<AiTaskMapper, AiTask> impleme
}
}
private Map<String, Long> buildAsrQueueCountMap(List<AiTask> tasks, Map<Long, AiModelVO> asrModelCache) {
Map<String, Long> 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<Long, AiModelVO> 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<Long, AiModelVO> 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;

57
backend/src/main/java/com/imeeting/support/redis/MeetingAsrPermitCache.java
View File

@ -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();
}
}