From 4ddef78f0af0d3ac9da43db650b94ff64df4fad1 Mon Sep 17 00:00:00 2001 From: jack ning Date: Mon, 24 Nov 2025 15:43:56 +0800 Subject: [PATCH] update --- .../bytedesk/core/constant/RedisConsts.java | 5 + .../ConnectionHeartbeatFlushTask.java | 10 +- .../connection/ConnectionRestService.java | 10 +- .../WorkgroupThreadRoutingStrategy.java | 138 ++++++++++-------- 4 files changed, 92 insertions(+), 71 deletions(-) diff --git a/modules/core/src/main/java/com/bytedesk/core/constant/RedisConsts.java b/modules/core/src/main/java/com/bytedesk/core/constant/RedisConsts.java index 517efb067a..84e5925e7d 100644 --- a/modules/core/src/main/java/com/bytedesk/core/constant/RedisConsts.java +++ b/modules/core/src/main/java/com/bytedesk/core/constant/RedisConsts.java @@ -50,4 +50,9 @@ public class RedisConsts { // 验证码相关常量 public static final String KAPTCHA_PREFIX = BYTEDESK_REDIS_PREFIX + "kaptcha:"; + // Redis 缓存心跳Key + public static final String REDIS_HEARTBEAT_HASH_KEY = RedisConsts.BYTEDESK_REDIS_PREFIX + "core:conn:hb"; + // Redis 最近一次数据库写入时间Key + public static final String REDIS_LAST_DB_WRITE_HASH_KEY = RedisConsts.BYTEDESK_REDIS_PREFIX + "core:conn:hb:lastdb"; + } diff --git a/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionHeartbeatFlushTask.java b/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionHeartbeatFlushTask.java index 6f1a92a54d..933b709394 100644 --- a/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionHeartbeatFlushTask.java +++ b/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionHeartbeatFlushTask.java @@ -25,9 +25,7 @@ public class ConnectionHeartbeatFlushTask { private final StringRedisTemplate stringRedisTemplate; private final ConnectionRestService connectionRestService; - private static final String REDIS_HEARTBEAT_HASH_KEY = RedisConsts.BYTEDESK_REDIS_PREFIX + "core:conn:hb"; - private static final String REDIS_LAST_DB_WRITE_HASH_KEY = RedisConsts.BYTEDESK_REDIS_PREFIX + "core:conn:hb:lastdb"; - + // 每 10 秒批量刷新一次 @Scheduled(fixedDelay = 10_000) @Transactional @@ -37,7 +35,7 @@ public class ConnectionHeartbeatFlushTask { } try { long start = System.nanoTime(); - Map entries = stringRedisTemplate.opsForHash().entries(REDIS_HEARTBEAT_HASH_KEY); + Map entries = stringRedisTemplate.opsForHash().entries(RedisConsts.REDIS_HEARTBEAT_HASH_KEY); if (entries == null || entries.isEmpty()) { return; } @@ -62,12 +60,12 @@ public class ConnectionHeartbeatFlushTask { for (Map.Entry e : heartbeats.entrySet()) { String clientId = e.getKey(); Long hbTs = e.getValue(); - Object lastDbStr = stringRedisTemplate.opsForHash().get(REDIS_LAST_DB_WRITE_HASH_KEY, clientId); + Object lastDbStr = stringRedisTemplate.opsForHash().get(RedisConsts.REDIS_LAST_DB_WRITE_HASH_KEY, clientId); if (lastDbStr != null) { try { long lastDb = Long.parseLong(String.valueOf(lastDbStr)); if (lastDb >= hbTs) { - stringRedisTemplate.opsForHash().delete(REDIS_HEARTBEAT_HASH_KEY, clientId); + stringRedisTemplate.opsForHash().delete(RedisConsts.REDIS_HEARTBEAT_HASH_KEY, clientId); } } catch (NumberFormatException ignore) {} } diff --git a/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionRestService.java b/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionRestService.java index 0bf36ac081..19847d04f3 100644 --- a/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionRestService.java +++ b/modules/core/src/main/java/com/bytedesk/core/socket/connection/ConnectionRestService.java @@ -28,6 +28,7 @@ import org.springframework.transaction.annotation.Transactional; import org.springframework.util.StringUtils; import org.springframework.data.redis.core.StringRedisTemplate; import com.bytedesk.core.base.BaseRestServiceWithExport; +import com.bytedesk.core.constant.RedisConsts; import com.bytedesk.core.rbac.auth.AuthService; import com.bytedesk.core.rbac.user.UserEntity; import com.bytedesk.core.uid.UidUtils; @@ -56,10 +57,7 @@ public class ConnectionRestService extends BaseRestServiceWithExport availableAgents = presenceFacadeService.getAvailableAgents(workgroup); + if (availableAgents.isEmpty()) { + log.info("无在线客服可用，降级为离线留言"); + return routeToOfflineMessage(visitorRequest, thread, workgroup); + } + long selectStartTime = System.currentTimeMillis(); - AgentEntity agentEntity = selectAgent(workgroup, thread); - log.info("客服选择完成 - agentUid: {}, 选择耗时: {}ms", agentEntity.getUid(), System.currentTimeMillis() - selectStartTime); + AgentEntity agentEntity = resolvePreferredAgent(workgroup, thread, availableAgents); + if (agentEntity == null) { + log.warn("未能根据路由策略选出可用客服，降级为离线留言 - workgroupUid: {}", workgroup.getUid()); + return routeToOfflineMessage(visitorRequest, thread, workgroup); + } + log.info("客服选择完成 - agentUid: {}, 选择耗时: {}ms", agentEntity.getUid(), + System.currentTimeMillis() - selectStartTime); // 加入队列 log.debug("开始将线程加入工作组队列"); long enqueueStartTime = System.currentTimeMillis(); QueueService.QueueEnqueueResult enqueueResult = queueService - .enqueueWorkgroupWithResult(thread, agentEntity, workgroup, visitorRequest); + .enqueueWorkgroupWithResult(thread, agentEntity, workgroup, visitorRequest); QueueMemberEntity queueMemberEntity = enqueueResult.queueMember(); - log.info("工作组队列加入完成 - queueMemberUid: {}, 耗时: {}ms", queueMemberEntity.getUid(), System.currentTimeMillis() - enqueueStartTime); + log.info("工作组队列加入完成 - queueMemberUid: {}, 耗时: {}ms", queueMemberEntity.getUid(), + System.currentTimeMillis() - enqueueStartTime); - // 处理强制转人工 if (visitorRequest.getForceAgent()) { log.debug("处理强制转人工标记"); handleForceAgentTransfer(visitorRequest, thread, queueMemberEntity); } - // 根据客服状态进行路由 - log.debug("开始根据客服状态进行最终路由"); - return routeByAgentStatus(agentEntity, thread, queueMemberEntity, workgroup, visitorRequest); + boolean onlineAndAvailable = presenceFacadeService.isAgentOnlineAndAvailable(agentEntity); + if (onlineAndAvailable) { + log.debug("客服在线且可接待，检查接待名额"); + boolean hasCapacity = queueMemberEntity.getWorkgroupQueue().getChattingCount() < agentEntity.getMaxThreadCount(); + if (hasCapacity) { + log.info("客服有接待名额，进入接待流程 - agentUid: {}, agentName {}", agentEntity.getUid(), + agentEntity.getNickname()); + return handleAvailableWorkgroup(thread, agentEntity, queueMemberEntity); + } + log.info("客服接待名额已满，进入排队 - agentUid: {}, agentName {}", agentEntity.getUid(), + agentEntity.getNickname()); + return handleQueuedWorkgroup(thread, agentEntity, queueMemberEntity); + } + + log.info("客服离线或不可接待，降级为离线留言 - agentUid: {}, agentName {}", agentEntity.getUid(), + agentEntity.getNickname()); + return getOfflineMessage(visitorRequest, thread, agentEntity, workgroup, queueMemberEntity); } /** - * 选择客服 - * TODO: 存在选择逻辑bug：当某个客服agent在线，却选择了另外不在线的客服agent，导致无法正确路由，请首先排除离线客服agent， - * 请结合在线状态和接待负载进行优化 - * 合并 selectAgent 和 routeByAgentStatus 方法 + * 选择优先客服：优先使用路由策略结果，其次选择第一个在线可用客服 */ - private AgentEntity selectAgent(WorkgroupEntity workgroup, ThreadEntity thread) { - AgentEntity agentEntity = workgroupRoutingService.selectAgent(workgroup, thread); + private AgentEntity resolvePreferredAgent(WorkgroupEntity workgroup, ThreadEntity thread, + List candidates) { + if (candidates == null || candidates.isEmpty()) { + return null; + } + log.debug("选择优先客服：优先使用路由策略结果，其次选择第一个在线可用客服"); - if (agentEntity == null) { - // 离线留言接待客服 - agentEntity = workgroup.getMessageLeaveAgent(); - if (agentEntity == null) { - log.error("离线留言接待客服不存在，请配置工作组留言接待客服"); - throw new IllegalStateException("Workgroup message leave agent not found"); + AgentEntity routedAgent = workgroupRoutingService.selectAgent(workgroup, thread); + if (routedAgent != null && presenceFacadeService.isAgentOnlineAndAvailable(routedAgent)) { + boolean existsInCandidate = candidates.stream() + .anyMatch(agent -> StringUtils.hasText(agent.getUid()) + && agent.getUid().equals(routedAgent.getUid())); + if (existsInCandidate) { + return routedAgent; } } + log.debug("未能使用路由策略结果，选择第一个在线可用客服"); - return agentEntity; + return candidates.stream() + .filter(presenceFacadeService::isAgentOnlineAndAvailable) + .findFirst() + .orElse(null); + } + + /** + * 不满足服务时间或无在线客服时，统一进入离线留言流程 + */ + private MessageProtobuf routeToOfflineMessage(VisitorRequest visitorRequest, ThreadEntity thread, + WorkgroupEntity workgroup) { + AgentEntity messageLeaveAgent = workgroup.getMessageLeaveAgent(); + if (messageLeaveAgent == null) { + log.error("离线留言接待客服不存在，请配置工作组留言接待客服 - workgroupUid: {}", workgroup.getUid()); + throw new IllegalStateException("Workgroup message leave agent not found"); + } + log.debug("使用离线留言接待客服 - agentUid: {}", messageLeaveAgent.getUid()); + QueueMemberEntity queueMemberEntity = queueService + .enqueueWorkgroupWithResult(thread, messageLeaveAgent, workgroup, visitorRequest) + .queueMember(); + return getOfflineMessage(visitorRequest, thread, messageLeaveAgent, workgroup, queueMemberEntity); } /** @@ -470,29 +513,6 @@ public class WorkgroupThreadRoutingStrategy extends AbstractThreadRoutingStrateg } } - /** - * 根据客服状态进行路由 - */ - private MessageProtobuf routeByAgentStatus(AgentEntity agentEntity, ThreadEntity thread, - QueueMemberEntity queueMemberEntity, WorkgroupEntity workgroup, VisitorRequest visitorRequest) { - - if (presenceFacadeService.isAgentOnlineAndAvailable(agentEntity)) { - log.info("客服在线且可接待 - agentUid: {}, agentName {}", agentEntity.getUid(), agentEntity.getNickname()); - // 客服在线且可接待 - if (queueMemberEntity.getWorkgroupQueue().getChattingCount() < agentEntity.getMaxThreadCount()) { - log.info("客服有可用接待名额 - agentUid: {}, agentName {}", agentEntity.getUid(), agentEntity.getNickname()); - // 有可用接待名额 - return handleAvailableWorkgroup(thread, agentEntity, queueMemberEntity); - } else { - log.info("客服接待名额已满，进入排队 - agentUid: {}, agentName {}", agentEntity.getUid(), agentEntity.getNickname()); - return handleQueuedWorkgroup(thread, agentEntity, queueMemberEntity); - } - } else { - log.info("客服离线或不可接待 - agentUid: {}, agentName {}", agentEntity.getUid(), agentEntity.getNickname()); - // 客服离线或不可接待 - return getOfflineMessage(visitorRequest, thread, agentEntity, workgroup, queueMemberEntity); - } - } /** * 处理可用工作组客服