From 7123da8cf9bf283b96ffaecffdb3bbaed0db7433 Mon Sep 17 00:00:00 2001 From: wangxiang <1827945911@qq.com> Date: Fri, 16 Sep 2022 01:26:22 +0800 Subject: [PATCH] =?UTF-8?q?:rocket:=20=E9=9B=86=E6=88=90SSE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/impl/MapLogisticSseServiceImpl.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/kicc-platform/kicc-platform-biz/kicc-common-biz/src/main/java/com/cloud/kicc/commonbiz/service/impl/MapLogisticSseServiceImpl.java b/kicc-platform/kicc-platform-biz/kicc-common-biz/src/main/java/com/cloud/kicc/commonbiz/service/impl/MapLogisticSseServiceImpl.java index 4b1245d1..968edfdf 100644 --- a/kicc-platform/kicc-platform-biz/kicc-common-biz/src/main/java/com/cloud/kicc/commonbiz/service/impl/MapLogisticSseServiceImpl.java +++ b/kicc-platform/kicc-platform-biz/kicc-common-biz/src/main/java/com/cloud/kicc/commonbiz/service/impl/MapLogisticSseServiceImpl.java @@ -14,6 +14,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import java.io.IOException; import java.time.LocalDateTime; import java.util.Iterator; import java.util.Optional; @@ -70,10 +71,7 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService { log.info("clientId:{},用户Id为:{}的SSE长轮询已经返回响应关闭,正在删除当前的建立通道对象", clientId, kiccUser.getId()); sseSignalContainers.removeIf(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), kiccUser.getTenantId())); }); - sseEmitter.onError(e -> { - log.info("clientId:{},当前用户Id为:{}的SSE长轮询出现异常,正在删除当前的建立通道对象,错误信息{}", clientId, kiccUser.getId(), e.getLocalizedMessage()); - sseSignalContainers.removeIf(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), kiccUser.getTenantId())); - }); + sseEmitter.onError(e -> log.info("clientId:{},当前用户Id为:{}的SSE长轮询出现异常,正在删除当前的建立通道对象,错误信息{}", clientId, kiccUser.getId(), e.getLocalizedMessage())); return sseEmitter; } @@ -129,12 +127,17 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService { /** * 执行心跳维护,避免 sse 膨胀容量问题 */ - @SneakyThrows private void doMaintenance() { Iterator it = sseSignalContainers.iterator(); while (it.hasNext()) { SseSignalContainer item = it.next(); - item.getSseEmitter().send(SseEmitter.event().comment("保持心跳" + LocalDateTime.now()).reconnectTime(1000)); + try { + item.getSseEmitter().send(SseEmitter.event().comment("保持心跳" + LocalDateTime.now()).reconnectTime(1000)); + } catch (IOException e) { + log.debug("clientId:{},用户Id为:{}发送心跳包失败", item.getClientId(), item.getUserId()); + sseSignalContainers.removeIf(sseSignalContainer -> + StrUtil.equals(sseSignalContainer.getClientId(), item.getClientId()) && StrUtil.equals(sseSignalContainer.getTenantId(), item.getTenantId())); + } } }