Browse Source

🚀 集成SSE

master
wangxiang 3 years ago
parent
commit
7123da8cf9
  1. 13
      kicc-platform/kicc-platform-biz/kicc-common-biz/src/main/java/com/cloud/kicc/commonbiz/service/impl/MapLogisticSseServiceImpl.java

13
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.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.Iterator; import java.util.Iterator;
import java.util.Optional; import java.util.Optional;
@ -70,10 +71,7 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService {
log.info("clientId:{},用户Id为:{}的SSE长轮询已经返回响应关闭,正在删除当前的建立通道对象", clientId, kiccUser.getId()); log.info("clientId:{},用户Id为:{}的SSE长轮询已经返回响应关闭,正在删除当前的建立通道对象", clientId, kiccUser.getId());
sseSignalContainers.removeIf(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), kiccUser.getTenantId())); sseSignalContainers.removeIf(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), kiccUser.getTenantId()));
}); });
sseEmitter.onError(e -> { sseEmitter.onError(e -> log.info("clientId:{},当前用户Id为:{}的SSE长轮询出现异常,正在删除当前的建立通道对象,错误信息{}", clientId, kiccUser.getId(), e.getLocalizedMessage()));
log.info("clientId:{},当前用户Id为:{}的SSE长轮询出现异常,正在删除当前的建立通道对象,错误信息{}", clientId, kiccUser.getId(), e.getLocalizedMessage());
sseSignalContainers.removeIf(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), kiccUser.getTenantId()));
});
return sseEmitter; return sseEmitter;
} }
@ -129,12 +127,17 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService {
/** /**
* 执行心跳维护,避免 sse 膨胀容量问题 * 执行心跳维护,避免 sse 膨胀容量问题
*/ */
@SneakyThrows
private void doMaintenance() { private void doMaintenance() {
Iterator<SseSignalContainer> it = sseSignalContainers.iterator(); Iterator<SseSignalContainer> it = sseSignalContainers.iterator();
while (it.hasNext()) { while (it.hasNext()) {
SseSignalContainer item = it.next(); SseSignalContainer item = it.next();
try {
item.getSseEmitter().send(SseEmitter.event().comment("保持心跳" + LocalDateTime.now()).reconnectTime(1000)); 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()));
}
} }
} }

Loading…
Cancel
Save