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

refactor(android): 重构Android gRPC服务与实时会议缓存 

将AndroidRealtimeMeetingGrpcService中的逻辑剥离，增强RealtimeMeetingSessionCache与RedisSupport的缓存支持，优化Android端推送与实时会议状态管理。
dev_na
chenhao 2026-09-18 14:07:32 +08:00
parent 58d647d8c8
commit 9f0a4f8c0a
12 changed files with 274 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -107,6 +107,14 @@ public final class RedisKeys {
return "biz:meeting:realtime:publisher:" + meetingId;
}
public static String realtimeMeetingLiveMeetingsKey() {
return "biz:meeting:realtime:live-meetings";
}
public static String realtimeMeetingPublisherDeviceKey(Long meetingId) {
return "biz:meeting:realtime:publisher-device:" + meetingId;
}
public static String realtimeMeetingResumeTimeoutPrefix() {
return "biz:meeting:realtime:resume-timeout:";
}

25
backend/src/main/java/com/imeeting/grpc/push/AndroidPushGrpcService.java
View File

@ -7,6 +7,7 @@ import com.imeeting.service.android.AndroidDeviceSessionService;
import com.imeeting.service.android.AndroidGatewayPushService;
import com.imeeting.service.android.AndroidPushMessageService;
import com.imeeting.service.biz.DeviceOnlineManagementService;
import com.imeeting.service.realtime.RealtimeMeetingPushSubscriptionManager;
import com.unisbase.common.exception.BusinessException;
import io.grpc.BindableService;
import io.grpc.stub.StreamObserver;
@ -29,6 +30,7 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase
private final AndroidGatewayPushService androidGatewayPushService;
private final AndroidPushMessageService androidPushMessageService;
private final DeviceOnlineManagementService deviceOnlineManagementService;
private final RealtimeMeetingPushSubscriptionManager realtimeMeetingPushSubscriptionManager;
@Override
public StreamObserver<ClientMessage> communicate(StreamObserver<ServerMessage> responseObserver) {
@ -151,6 +153,20 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase
.setMessage(connectionId)
.build())
.build());
attachRealtimePushSubscriptions(authContext);
}
/**
* 推送长连接即实时会议订阅入口，只挂接该用户作为创建人或参会人的进行中实时会议。
*/
private void attachRealtimePushSubscriptions(AndroidAuthContext authContext) {
try {
realtimeMeetingPushSubscriptionManager.onPushConnectionOpened(
authContext.getTenantId(), authContext.getUserId(), deviceId, connectionId);
} catch (Exception ex) {
log.warn("Failed to attach realtime push subscriptions, connectionId={}, deviceId={}",
connectionId, deviceId, ex);
}
}
private void handleHeartbeat(HeartbeatRequest request) {
@ -208,6 +224,7 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase
return;
}
AndroidDeviceSessionState state = androidDeviceSessionService.getByConnectionId(connectionId);
releaseRealtimePushSubscriptions(connectionId);
androidGatewayPushService.unregister(connectionId);
androidDeviceSessionService.closeSession(connectionId);
deviceOnlineManagementService.recordDisconnected(deviceId, state == null ? null : state.getLastSeenAt());
@ -217,6 +234,14 @@ public class AndroidPushGrpcService extends PushServiceGrpc.PushServiceImplBase
platform = null;
connected = false;
}
private void releaseRealtimePushSubscriptions(String targetConnectionId) {
try {
realtimeMeetingPushSubscriptionManager.onPushConnectionClosed(targetConnectionId);
} catch (Exception ex) {
log.warn("Failed to release realtime push subscriptions, connectionId={}", targetConnectionId, ex);
}
}
};
}

138
backend/src/main/java/com/imeeting/grpc/realtime/AndroidRealtimeMeetingGrpcService.java
View File

@ -53,7 +53,6 @@ public class AndroidRealtimeMeetingGrpcService
private final RealtimeMeetingEventRelay eventRelay;
private final RealtimeMeetingPushSubscriptionManager pushSubscriptionManager;
private final Map<Long, GrpcSession> sessions = new ConcurrentHashMap<>();
private final Map<String, GrpcSession> subscriberSessions = new ConcurrentHashMap<>();
@Override
public StreamObserver<ClientMessage> stream(StreamObserver<ServerMessage> responseObserver) {
@ -178,8 +177,8 @@ public class AndroidRealtimeMeetingGrpcService
session.context.getChannelState().put("modelCode", model.getModelCode());
session.context.getChannelState().put("mediaConfig", model.getMediaConfig());
session.context.setCallback(new GrpcChannelCallback(session));
detachSubscriber(session);
session.publisher = true;
pushSubscriptionManager.markPublisherDevice(session.meetingId, session.deviceId);
session.cleanupAction = this::cleanup;
try {
RealtimeMeetingMetrics.publisherOpened("grpc");
@ -253,9 +252,7 @@ public class AndroidRealtimeMeetingGrpcService
if (session.publisher) {
publisherLease.release(session.meetingId, session.publisherFenceToken);
RealtimeMeetingMetrics.publisherClosed("grpc");
}
if (!session.publisher) {
detachSubscriber(session);
pushSubscriptionManager.clearPublisherDevice(session.meetingId, session.deviceId);
}
if (session.publisher) {
sessionStateService.pauseByDisconnect(session.meetingId, session.publisherFenceToken);
@ -265,14 +262,6 @@ public class AndroidRealtimeMeetingGrpcService
}
}
private void detachSubscriber(GrpcSession target) {
if (target.pushSubscribed) {
target.pushSubscribed = false;
pushSubscriptionManager.unsubscribe(target.connectionId);
}
subscriberSessions.remove(target.connectionId, target);
}
private GrpcSession open(ConnectRequest connect, StreamObserver<ServerMessage> observer) throws Exception {
if (connect.getDeviceId().isBlank()) {
fail("REALTIME_DEVICE_ID_REQUIRED", "device_id 不能为空", false);
@ -295,122 +284,29 @@ public class AndroidRealtimeMeetingGrpcService
AndroidAuthContext authContext = androidAuthService.authenticateGrpc(
connect.getDeviceId(), connect.getAppVersion(), "android",
connect.getUserId(), connect.getTenantId());
boolean publisher = false;
boolean canPublish = authContext.getUserId() != null && authContext.getUserId().equals(meeting.getCreatorId());
if (publisher) {
if (authContext.getUserId() == null || !authContext.getUserId().equals(meeting.getCreatorId())) {
throw new SecurityException("只有会议创建人可以发布实时音频");
}
meetingAuthorizationService.assertCanControlRealtimeMeeting(
meeting, authContext, MeetingTerminalEnum.CUSTOM_TERMINAL.getCode());
} else {
meetingAuthorizationService.assertCanViewMeeting(meeting, authContext);
RealtimeMeetingSessionStatusVO currentStatus = sessionStateService.getStatus(meetingId);
if ((currentStatus == null || !"ACTIVE".equals(currentStatus.getStatus())) && !canPublish) {
throw new IllegalStateException("会议尚未开始，暂不能订阅实时转写");
}
if (authContext.getUserId() == null || !authContext.getUserId().equals(meeting.getCreatorId())) {
throw new SecurityException("实时会议 gRPC 只接受发布端连接，订阅方请使用推送长连接");
}
meetingAuthorizationService.assertCanControlRealtimeMeeting(
meeting, authContext, MeetingTerminalEnum.CUSTOM_TERMINAL.getCode());
if (!MeetingConstants.TYPE_REALTIME.equalsIgnoreCase(meeting.getMeetingType())) {
throw new IllegalStateException("当前会议不是实时会议");
}
sessionStateService.initSessionIfAbsent(meetingId, meeting.getTenantId(), authContext.getUserId());
if (publisher) {
sessionStateService.assertCanOpenSession(meetingId);
}
String connectionId = connect.getConnectionId().isBlank()
? "grpc-" + java.util.UUID.randomUUID().toString().replace("-", "")
: connect.getConnectionId().trim();
if (!publisher) {
GrpcSession next = new GrpcSession(meetingId, connectionId, null, observer, null);
next.publisher = false;
next.canPublish = canPublish;
String subscriberDeviceId = connect.getDeviceId().trim();
String rejectReason = pushSubscriptionManager.subscribe(
meetingId, connectionId, subscriberDeviceId, meeting.getTitle());
if (rejectReason != null) {
throw new IllegalStateException(rejectReason);
}
next.pushSubscribed = true;
try {
observer.onNext(ServerMessage.newBuilder()
.setConnectAck(ConnectResponse.newBuilder()
.setSuccess(true)
.setMessage("实时转写订阅成功，转写经推送通道下发")
.build())
.build());
pushSubscriptionManager.replayHistory(meetingId, subscriberDeviceId, meeting.getTitle());
subscriberSessions.put(connectionId, next);
return next;
} catch (RuntimeException ex) {
detachSubscriber(next);
throw ex;
}
}
RealtimeMeetingSessionStatusVO status = sessionStateService.getStatus(meetingId);
RealtimeMeetingResumeConfig config = status == null ? null : status.getResumeConfig();
if (config == null || config.getAsrModelId() == null) {
throw new IllegalStateException("实时会议缺少 ASR 运行参数");
}
AiModelVO model = aiModelService.getModelById(config.getAsrModelId(), "ASR");
if (model == null) {
throw new IllegalStateException("实时 ASR 模型不存在");
}
RealtimeAsrChannel channel = channelFactory.getRequired(model.getProvider());
String publisherFenceToken = java.util.UUID.randomUUID().toString();
GrpcSession next = new GrpcSession(meetingId, connectionId, channel, observer, publisherFenceToken);
next.publisher = publisher;
next.context.setMeetingId(meetingId);
next.context.setProvider(channelFactory.normalizeProvider(model.getProvider()));
next.context.setTargetWsUrl(channel.resolveTargetWsUrl(model));
next.context.bindTransport(connectionId);
next.context.setConnectionId(publisherFenceToken);
next.context.getChannelState().put("modelCode", model.getModelCode());
next.context.getChannelState().put("mediaConfig", model.getMediaConfig());
next.context.setCallback(new GrpcChannelCallback(next));
if (!publisherLease.tryAcquire(meetingId, publisherFenceToken)) {
RealtimeMeetingMetrics.publisherRejected("grpc");
throw new IllegalStateException("当前会议已有活跃发布连接");
}
if (!sessionStateService.activate(meetingId, publisherFenceToken)) {
publisherLease.release(meetingId, publisherFenceToken);
throw new IllegalStateException("当前会议已有活跃连接");
}
GrpcSession previous = sessions.putIfAbsent(meetingId, next);
if (previous != null) {
publisherLease.release(meetingId, publisherFenceToken);
sessionStateService.pauseByDisconnect(meetingId, publisherFenceToken);
throw new IllegalStateException("当前会议已有活跃连接");
}
try {
session = next;
next.cleanupAction = this::cleanup;
RealtimeMeetingMetrics.publisherOpened("grpc");
audioStorageService.openSession(meetingId, connectionId);
channel.connect(next.context);
String start = OBJECT_MAPPER.writeValueAsString(channel.buildStartMessage(
model, config.getMode(), config.getLanguage(), config.getUseSpkId(),
config.getEnablePunctuation(), config.getEnableItn(), config.getEnableTextRefine(),
config.getSaveAudio(), config.getHotwords()));
channel.handleFrontendText(next.context, start);
observer.onNext(ServerMessage.newBuilder()
.setConnectAck(ConnectResponse.newBuilder().setSuccess(true).setMessage("实时会议连接成功"))
.build());
return next;
} catch (Exception ex) {
sessions.remove(meetingId, next);
publisherLease.release(meetingId, publisherFenceToken);
try {
channel.closeMeeting(next.context);
} catch (Exception closeEx) {
log.debug("Failed to close realtime upstream after handshake failure", closeEx);
}
next.context.closeTransport();
audioStorageService.closeSession(connectionId);
sessionStateService.pauseByDisconnect(meetingId, publisherFenceToken);
throw ex;
}
GrpcSession next = new GrpcSession(meetingId, connectionId, null, observer, null);
next.deviceId = connect.getDeviceId().trim();
next.publisher = false;
next.canPublish = true;
session = next;
next.cleanupAction = this::cleanup;
observer.onNext(ServerMessage.newBuilder()
.setConnectAck(ConnectResponse.newBuilder().setSuccess(true).setMessage("实时会议连接成功"))
.build());
return next;
}
};
}
@ -449,8 +345,8 @@ public class AndroidRealtimeMeetingGrpcService
private volatile boolean cleaned;
private volatile boolean publisher;
private boolean canPublish;
private String deviceId;
private Runnable cleanupAction;
private volatile boolean pushSubscribed;
private GrpcSession(Long meetingId, String connectionId, RealtimeAsrChannel channel,
StreamObserver<ServerMessage> observer, String publisherFenceToken) {

8
backend/src/main/java/com/imeeting/service/android/AndroidGatewayPushService.java
View File

@ -1,10 +1,13 @@
package com.imeeting.service.android;
import com.imeeting.dto.android.AndroidGrpcConnectionSnapshotVO;
import com.imeeting.dto.android.AndroidGrpcConnectionDetailVO;
import com.imeeting.grpc.push.PushMessage;
import com.imeeting.grpc.push.ServerMessage;
import io.grpc.stub.StreamObserver;
import java.util.List;
public interface AndroidGatewayPushService {
String register(String connectionId,
String deviceId,
@ -20,6 +23,11 @@ public interface AndroidGatewayPushService {
int pushToUser(Long tenantId, Long userId, PushMessage message);
/**
* 查询某用户在本实例上的全部推送长连接，只按 user_id 命中，不依赖设备授权租户；同一用户的多设备连接全部返回。
*/
List<AndroidGrpcConnectionDetailVO> listUserConnections(Long userId);
String disconnectDevice(String deviceId);
AndroidGrpcConnectionSnapshotVO snapshotConnections();

33
backend/src/main/java/com/imeeting/service/android/impl/AndroidAuthServiceImpl.java
View File

@ -50,22 +50,31 @@ public class AndroidAuthServiceImpl implements AndroidAuthService {
assertDeviceEnabled(device);
Long requestedUserId = parseOptionalLong(userId, "Android gRPC userId");
Long requestedTenantId = parseOptionalLong(tenantId, "Android gRPC tenantId");
boolean anonymous = requestedUserId == null;
AndroidAuthContext context = buildContext(anonymous ? "NONE" : "GRPC_USER", anonymous, deviceId, null, appVersion, platform, null, null, null, null);
Long resolvedTenantId = requestedTenantId != null ? requestedTenantId : license.getTenantId();
if (requestedUserId != null) {
if (requestedTenantId != null && !requestedTenantId.equals(license.getTenantId())) {
throw new RuntimeException("登录租户与授权租户不一致,无法绑定grpc");
}
context.setUserId(requestedUserId);
context.setTenantId(resolvedTenantId);
return context;
if (requestedUserId != null && requestedTenantId != null && !requestedTenantId.equals(license.getTenantId())) {
throw new RuntimeException("登录租户与授权租户不一致,无法绑定grpc");
}
context.setUserId(null);
context.setTenantId(resolvedTenantId);
// 客户端未带 user_id 时回落到设备绑定用户，推送长连接必须具备用户身份，按用户下发的推送才能命中。
Long resolvedUserId = requestedUserId != null ? requestedUserId : device.getUserId();
boolean anonymous = resolvedUserId == null;
AndroidAuthContext context = buildContext(anonymous ? "NONE" : "GRPC_USER", anonymous, deviceId, null, appVersion, platform, null, null, null, null);
context.setUserId(resolvedUserId);
context.setTenantId(resolveGrpcTenantId(requestedTenantId, resolvedUserId, device, license));
return context;
}
/**
* 推送长连接的租户优先级：客户端显式租户、绑定用户所属设备租户、设备授权租户。
*/
private Long resolveGrpcTenantId(Long requestedTenantId, Long resolvedUserId, DeviceInfoEntity device, LicenseEntity license) {
if (requestedTenantId != null) {
return requestedTenantId;
}
if (resolvedUserId != null && resolvedUserId.equals(device.getUserId()) && device.getTenantId() != null) {
return device.getTenantId();
}
return license.getTenantId();
}
@Override
public AndroidAuthContext authenticateRealtimeGrpc(String deviceId, String appVersion, String platform,
String userId, String tenantId, String accessToken) {

26
backend/src/main/java/com/imeeting/service/android/impl/AndroidGatewayPushServiceImpl.java
View File

@ -21,6 +21,7 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService
private final Map<String, Binding> byConnectionId = new ConcurrentHashMap<>();
private final Map<String, String> connectionByDeviceId = new ConcurrentHashMap<>();
private final Map<String, Map<String, Binding>> connectionsByUserKey = new ConcurrentHashMap<>();
private final Map<Long, Map<String, Binding>> connectionsByUserId = new ConcurrentHashMap<>();
@Override
public String register(String connectionId,
@ -103,6 +104,20 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService
return successCount;
}
@Override
public List<AndroidGrpcConnectionDetailVO> listUserConnections(Long userId) {
if (userId == null) {
return List.of();
}
Map<String, Binding> bindings = connectionsByUserId.get(userId);
if (bindings == null || bindings.isEmpty()) {
return List.of();
}
return bindings.entrySet().stream()
.map(entry -> toDetail(entry.getKey(), entry.getValue()))
.toList();
}
@Override
public String disconnectDevice(String deviceId) {
String connectionId = connectionByDeviceId.get(deviceId);
@ -142,6 +157,11 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService
}
private void addUserIndex(String connectionId, Binding binding) {
if (binding.userId() != null) {
connectionsByUserId
.computeIfAbsent(binding.userId(), ignored -> new ConcurrentHashMap<>())
.put(connectionId, binding);
}
String userKey = buildUserKey(binding.tenantId(), binding.userId());
if (userKey == null) {
return;
@ -156,6 +176,12 @@ public class AndroidGatewayPushServiceImpl implements AndroidGatewayPushService
}
private void removeUserIndex(String connectionId, Binding binding) {
if (binding.userId() != null) {
connectionsByUserId.computeIfPresent(binding.userId(), (ignored, bindings) -> {
bindings.remove(connectionId);
return bindings.isEmpty() ? null : bindings;
});
}
String userKey = buildUserKey(binding.tenantId(), binding.userId());
if (userKey == null) {
return;

6
backend/src/main/java/com/imeeting/service/biz/impl/DeviceOnlineManagementServiceImpl.java
View File

@ -69,8 +69,10 @@ public class DeviceOnlineManagementServiceImpl implements DeviceOnlineManagement
// existing.setTerminalType(normalizeTerminalType(authContext.getPlatform()));
existing.setTerminalVersion(normalize(authContext.getAppVersion()));
existing.setLastOnlineAt(now);
existing.setUserId(authContext.getUserId());
existing.setTenantId(authContext.getTenantId());
// 匿名连接只刷新在线信息，保留既有绑定用户，否则按用户下发的实时推送会全部失效。
if (authContext.getTenantId() != null) {
existing.setTenantId(authContext.getTenantId());
}
deviceInfoMapper.updateConnectionInfoByIdIgnoreTenant(existing);
}

40
backend/src/main/java/com/imeeting/service/biz/impl/MeetingCommandServiceImpl.java
View File

@ -472,8 +472,8 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
long transcriptCount = transcriptMapper.selectCount(new LambdaQueryWrapper<MeetingTranscript>()
.eq(MeetingTranscript::getMeetingId, meetingId));
if (transcriptCount <= 0) {
realtimeMeetingSessionStateService.pause(meetingId);
throw new RuntimeException("当前还没有转录内容，无法结束会议。请先开始识别，或直接离开页面稍后继续。");
completeRealtimeMeetingWithoutTranscript(meeting, currentStatus, audioUrl);
return;
}
if ((audioUrl == null || audioUrl.isBlank()) && (meeting.getAudioUrl() == null || meeting.getAudioUrl().isBlank())) {
@ -497,6 +497,42 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
}
}
/**
* 没有最终转写时的完成处理：有录音就转离线识别，既无录音也无转写时按已完成结束，避免会议一直停留在录音中。
*/
private void completeRealtimeMeetingWithoutTranscript(Meeting meeting,
RealtimeMeetingSessionStatusVO currentStatus,
String audioUrl) {
Long meetingId = meeting.getId();
boolean audioProvided = (audioUrl != null && !audioUrl.isBlank())
|| (meeting.getAudioUrl() != null && !meeting.getAudioUrl().isBlank());
if (!audioProvided) {
applyRealtimeAudioFinalizeResult(meeting, realtimeMeetingAudioStorageService.finalizeMeetingAudio(meetingId));
if (meeting.getAudioUrl() == null || meeting.getAudioUrl().isBlank()) {
// 本来就没有录到音频，不属于音频保存失败，避免终端展示误导性的失败提示
meeting.setAudioSaveStatus(RealtimeMeetingAudioStorageService.STATUS_NONE);
meeting.setAudioSaveMessage(null);
}
}
RealtimeMeetingResumeConfig resumeConfig = currentStatus == null ? null : currentStatus.getResumeConfig();
if (meeting.getAudioUrl() != null && !meeting.getAudioUrl().isBlank()
&& resumeConfig != null && resumeConfig.getAsrModelId() != null) {
meetingService.updateById(meeting);
meetingDomainSupport.prewarmPlaybackAudioAfterCommit(meeting.getAudioUrl());
prepareOfflineReprocessTasks(meetingId, currentStatus);
realtimeMeetingSessionStateService.clear(meetingId);
updateMeetingProgress(meetingId, 0, "正在转入离线音频识别流程...", 0);
aiTaskService.triggerQueuedAsrScheduling();
return;
}
realtimeMeetingSessionStateService.clear(meetingId);
meeting.setStatus(MeetingStatusEnum.COMPLETED.getCode());
meetingService.updateById(meeting);
updateMeetingProgress(meetingId, 100, "未识别到有效录音数据，会议已结束", 0);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void finishOfflineMeeting(Long meetingId, String finishStage) {

8
backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java
View File

@ -8,6 +8,8 @@ import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheState;
import com.imeeting.entity.biz.Meeting;
import com.imeeting.entity.biz.MeetingTranscript;
import com.imeeting.enums.MeetingStatusEnum;
import com.imeeting.event.RealtimeMeetingSessionActivatedEvent;
import com.imeeting.event.RealtimeMeetingSessionClosedEvent;
import com.imeeting.mapper.biz.MeetingMapper;
import com.imeeting.mapper.biz.MeetingTranscriptMapper;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
@ -18,6 +20,7 @@ import com.imeeting.support.redis.RealtimeMeetingTranscriptCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import java.time.Duration;
@ -36,6 +39,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
private final MeetingMapper meetingMapper;
private final RealtimeMeetingTranscriptCache realtimeMeetingTranscriptCache;
private final RealtimeMeetingPublisherLease publisherLease;
private final ApplicationEventPublisher eventPublisher;
@Value("${imeeting.realtime.resume-window-minutes:30}")
private String resumeWindowMinutesValue;
@ -163,6 +167,8 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
sessionCache.clearResumeTimeout(meetingId);
sessionCache.clearEmptyTimeout(meetingId);
// 发布激活事件：推送订阅挂载由监听器完成，避免会话状态服务反向依赖推送层形成 Bean 循环依赖
eventPublisher.publishEvent(new RealtimeMeetingSessionActivatedEvent(meetingId));
return true;
}
@ -322,6 +328,8 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
@Override
public void clear(Long meetingId) {
sessionCache.clearAll(meetingId);
// 发布会话结束事件：推送订阅回收由监听器完成，避免会话状态服务反向依赖推送层形成 Bean 循环依赖
eventPublisher.publishEvent(new RealtimeMeetingSessionClosedEvent(meetingId));
}
private RealtimeMeetingSessionStatusVO pauseState(Long meetingId, RealtimeMeetingSessionState state) {

42
backend/src/main/java/com/imeeting/support/RedisSupport.java
View File

@ -11,6 +11,7 @@ import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
@Component
@Slf4j
@ -202,6 +203,47 @@ public class RedisSupport {
}
}
/**
* 写入短期成员，分数为 now+ttl，顺带清理已过期成员，用于跨实例可枚举的短期索引。
*/
public boolean touchExpiringMember(String key, String member, Duration ttl) {
if (key == null || key.isBlank() || member == null || member.isBlank() || ttl == null) {
return false;
}
try {
Long written = commands().eval(
"local now = tonumber(ARGV[1]); local ttl = tonumber(ARGV[2]); "
+ "redis.call('zremrangebyscore', KEYS[1], '-inf', now); "
+ "redis.call('zadd', KEYS[1], now + ttl, ARGV[3]); redis.call('pexpire', KEYS[1], ttl); return 1",
ScriptOutputType.INTEGER,
new String[]{key},
String.valueOf(System.currentTimeMillis()),
String.valueOf(ttl.toMillis()),
member
);
return written != null && written > 0;
} catch (Exception ex) {
log.warn("Touch Redis set member failed, key={}", key, ex);
return false;
}
}
/**
* 读取未过期的全部成员，读取前先剪掉已过期成员。
*/
public List<String> listUnexpiredMembers(String key) {
if (key == null || key.isBlank()) {
return List.of();
}
try {
commands().zremrangebyscore(key, Double.NEGATIVE_INFINITY, (double) System.currentTimeMillis());
List<String> members = commands().zrange(key, 0, -1);
return members == null ? List.of() : members;
} catch (Exception ex) {
log.warn("List Redis set members failed, key={}", key, ex);
return List.of();
}
}
public void removeFromSetQuietly(String key, String... members) {
if (members == null || members.length == 0) {
return;

63
backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingSessionCache.java
View File

@ -7,6 +7,7 @@ import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.List;
@Component
@RequiredArgsConstructor
@ -52,5 +53,67 @@ public class RealtimeMeetingSessionCache {
redisSupport.deleteQuietly(RedisKeys.realtimeMeetingResumeTimeoutKey(meetingId));
redisSupport.deleteQuietly(RedisKeys.realtimeMeetingEmptyTimeoutKey(meetingId));
redisSupport.deleteQuietly(RedisKeys.realtimeMeetingEventSeqKey(meetingId));
redisSupport.deleteQuietly(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId));
redisSupport.removeExpiringSlot(RedisKeys.realtimeMeetingLiveMeetingsKey(), String.valueOf(meetingId));
}
/**
* 登记进行中的实时会议，供其它实例发现并按参会人挂接推送订阅。
*/
public void markLive(Long meetingId, Duration ttl) {
if (meetingId == null) {
return;
}
redisSupport.touchExpiringMember(RedisKeys.realtimeMeetingLiveMeetingsKey(), String.valueOf(meetingId), ttl);
}
public void clearLive(Long meetingId) {
if (meetingId == null) {
return;
}
redisSupport.removeExpiringSlot(RedisKeys.realtimeMeetingLiveMeetingsKey(), String.valueOf(meetingId));
}
public List<Long> listLiveMeetingIds() {
return redisSupport.listUnexpiredMembers(RedisKeys.realtimeMeetingLiveMeetingsKey()).stream()
.map(RealtimeMeetingSessionCache::parseMeetingId)
.filter(RealtimeMeetingSessionCache::nonNullId)
.toList();
}
/**
* 记录正在发布音频的设备，跨实例抑制发布端重复收流。
*/
public void savePublisherDevice(Long meetingId, String deviceId, Duration ttl) {
if (meetingId == null || deviceId == null || deviceId.isBlank()) {
return;
}
redisSupport.setString(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId), deviceId.trim(), ttl);
}
public void clearPublisherDevice(Long meetingId, String deviceId) {
if (meetingId == null || deviceId == null || deviceId.isBlank()) {
return;
}
redisSupport.deleteIfValue(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId), deviceId.trim());
}
public String getPublisherDevice(Long meetingId) {
if (meetingId == null) {
return null;
}
return redisSupport.getStringQuietly(RedisKeys.realtimeMeetingPublisherDeviceKey(meetingId));
}
private static Long parseMeetingId(String value) {
try {
return Long.valueOf(value);
} catch (NumberFormatException ex) {
return null;
}
}
private static boolean nonNullId(Long meetingId) {
return meetingId != null;
}
}

29
frontend/src/pages/business/RealtimeAsrSession.tsx
View File

@ -31,6 +31,7 @@ import {
const SAMPLE_RATE = 16000;
const CHUNK_SIZE = 1280;
const CURRENT_PLATFORM = "WEB" as const;
const MEETING_STATUS_COMPLETED = 3;
type WsSpeaker = string | { name?: string; user_id?: string | number } | undefined;
type WsMessage = {
@ -884,31 +885,29 @@ export function RealtimeAsrSession() {
} catch {
// 会议完成已成功提交，详情刷新失败不应反向标记为结束失败。
}
// 没有转写时后端可能直接按已完成结束会议，不再进入总结流程。
const completedWithoutData = savedMeeting?.status === MEETING_STATUS_COMPLETED;
sessionStorage.removeItem(getSessionKey(meetingId));
setSessionStatus((prev) => prev ? { ...prev, status: "COMPLETING", canResume: false, activeConnection: false } : prev);
setStatusText("已提交总结任务");
setSessionStatus((prev) => prev ? {
...prev,
status: completedWithoutData ? "COMPLETED" : "COMPLETING",
canResume: false,
activeConnection: false,
} : prev);
setStatusText(completedWithoutData ? "未识别到有效录音数据，会议已结束" : "已提交总结任务");
if (savedMeeting?.audioSaveStatus === "FAILED") {
message.warning(savedMeeting.audioSaveMessage || "实时会议已完成，但音频保存失败，当前无法播放会议录音。转写和总结不受影响。");
} else if (completedWithoutData) {
message.success("未识别到有效录音数据，会议已结束");
} else {
message.success("实时会议已结束，正在生成总结");
}
if (navigateAfterStop) {
navigate(`/meetings/${meetingId}`);
}
} catch (error) {
} catch {
completeOnceRef.current = false;
const errorMessage = error instanceof Error ? error.message : "结束会议失败";
if (errorMessage.includes("当前还没有转录内容")) {
try {
const statusRes = await getRealtimeMeetingSessionStatus(meetingId);
setSessionStatus(statusRes.data.data);
} catch {
// ignore status refresh failure
}
setStatusText("当前还没有转录内容，可继续识别");
} else {
setStatusText("结束失败");
}
setStatusText("结束失败");
} finally {
setRecording(false);
setFinishing(false);