From 4d24075d226054bb901a2e2f68ef7c4744fead7f Mon Sep 17 00:00:00 2001 From: shenzhidan Date: Wed, 12 Aug 2026 15:04:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E4=BA=BA=E5=91=98?= =?UTF-8?q?=E5=AE=9A=E4=BD=8DWebSocket=E4=BC=81=E4=B8=9A=E4=B8=8E=E6=B8=AF?= =?UTF-8?q?=E5=8C=BA=E8=BF=87=E6=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- start/src/main/resources/kafka.yml | 4 +- .../MockPersonLocationPublisher.java | 5 +- .../PersonLocationHandshakeInterceptor.java | 95 +++++++ .../websocket/PersonLocationSubscription.java | 27 ++ .../PersonLocationWebSocketConfiguration.java | 2 + .../PersonLocationWebSocketHandler.java | 45 +++- .../query/FindsStaffMappingSyncService.java | 3 + .../query/RealtimePersonLocationMatcher.java | 22 +- .../kafka/LocationKafkaProbeListener.java | 33 ++- .../realtime/RealtimeLocationMessage.java | 25 ++ .../realtime/RealtimeLocationPublisher.java | 2 +- .../realtime/RealtimePersonRoutingCache.java | 245 ++++++++++++++++++ .../RealtimeSchedulingConfiguration.java | 9 + .../dataobject/RealtimePersonRoutingDO.java | 9 + .../mapper/PersonnelMatchMapper.java | 4 + .../repository/PersonnelMatchRepository.java | 3 + .../impl/PersonnelMatchRepositoryImpl.java | 4 + .../resources/mapper/PersonnelMatchMapper.xml | 32 +++ 18 files changed, 549 insertions(+), 20 deletions(-) create mode 100644 web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationHandshakeInterceptor.java create mode 100644 web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationSubscription.java create mode 100644 web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationMessage.java create mode 100644 web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimePersonRoutingCache.java create mode 100644 web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeSchedulingConfiguration.java create mode 100644 web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/dataobject/RealtimePersonRoutingDO.java diff --git a/start/src/main/resources/kafka.yml b/start/src/main/resources/kafka.yml index 89c3d62..3e2dd8f 100644 --- a/start/src/main/resources/kafka.yml +++ b/start/src/main/resources/kafka.yml @@ -21,9 +21,11 @@ personnel-positioning: location: enabled: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_ENABLED:true} topic: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_TOPIC:point_push} - # Each WebSocket-serving instance must consume every point so it can notify its own clients. + # 每个提供 WebSocket 的实例都要消费全部点位，以便向本实例连接的前端推送。 group-id: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_GROUP_ID:personnel-position-location-${HOSTNAME:standalone}} max-log-bytes: ${PERSONNEL_POSITIONING_KAFKA_LOCATION_MAX_LOG_BYTES:16384} + person-routing-cache-ttl-ms: ${PERSONNEL_POSITIONING_PERSON_ROUTING_CACHE_TTL_MS:300000} + person-routing-cache-check-ms: ${PERSONNEL_POSITIONING_PERSON_ROUTING_CACHE_CHECK_MS:30000} websocket: location: enabled: ${PERSONNEL_POSITIONING_WEBSOCKET_LOCATION_ENABLED:true} diff --git a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/MockPersonLocationPublisher.java b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/MockPersonLocationPublisher.java index f96c3b1..cd1a411 100644 --- a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/MockPersonLocationPublisher.java +++ b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/MockPersonLocationPublisher.java @@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import com.zcloud.personnel.positioning.dto.clientobject.BiPersonLocationCO; import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationPublisher; +import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.ObjectProvider; @@ -74,9 +75,11 @@ public class MockPersonLocationPublisher { BiPersonLocationCO location = createPerson(personIndex, trackIndex); String payload = serialize(location); + RealtimeLocationMessage message = new RealtimeLocationMessage( + payload, location.getCorpinfoId(), location.getPortArea()); int deliveries = 0; for (RealtimeLocationPublisher publisher : publishers) { - deliveries += publisher.publish(payload); + deliveries += publisher.publish(message); } LOGGER.debug("模拟人员点位已推送，staffNo={}, terminalNo={}, websocketDeliveries={}", location.getStaffNo(), location.getTerminalNo(), deliveries); diff --git a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationHandshakeInterceptor.java b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationHandshakeInterceptor.java new file mode 100644 index 0000000..da6b7ed --- /dev/null +++ b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationHandshakeInterceptor.java @@ -0,0 +1,95 @@ +package com.zcloud.personnel.positioning.websocket; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.http.HttpStatus; +import org.springframework.http.server.ServerHttpRequest; +import org.springframework.http.server.ServerHttpResponse; +import org.springframework.stereotype.Component; +import org.springframework.util.MultiValueMap; +import org.springframework.util.StringUtils; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.server.HandshakeInterceptor; +import org.springframework.web.util.UriComponentsBuilder; + +import java.util.List; +import java.util.Map; + +@Component +public class PersonLocationHandshakeInterceptor implements HandshakeInterceptor { + private static final Logger LOGGER = LoggerFactory.getLogger(PersonLocationHandshakeInterceptor.class); + + @Override + public boolean beforeHandshake(ServerHttpRequest request, + ServerHttpResponse response, + WebSocketHandler wsHandler, + Map attributes) { + try { + MultiValueMap params = UriComponentsBuilder.fromUri(request.getURI()) + .build().getQueryParams(); + Long corpinfoId = parseCorpinfoId(params); + Integer portArea = parsePortArea(params); + attributes.put(PersonLocationSubscription.SESSION_ATTRIBUTE, + new PersonLocationSubscription(corpinfoId, portArea)); + return true; + } catch (IllegalArgumentException e) { + response.setStatusCode(HttpStatus.BAD_REQUEST); + LOGGER.warn("Rejected personnel location WebSocket subscription, uri={}, reason={}", + request.getURI(), e.getMessage()); + return false; + } + } + + @Override + public void afterHandshake(ServerHttpRequest request, + ServerHttpResponse response, + WebSocketHandler wsHandler, + Exception exception) { + // 握手完成后无需额外处理。 + } + + private Long parseCorpinfoId(MultiValueMap params) { + if (!params.containsKey("corpinfoId")) { + return null; + } + String value = singleValue(params, "corpinfoId"); + if (!StringUtils.hasText(value)) { + return null; + } + try { + long corpinfoId = Long.parseLong(value.trim()); + if (corpinfoId <= 0L) { + throw new IllegalArgumentException("corpinfoId must be a positive integer"); + } + return corpinfoId; + } catch (NumberFormatException e) { + throw new IllegalArgumentException("corpinfoId must be a positive integer"); + } + } + + private Integer parsePortArea(MultiValueMap params) { + if (!params.containsKey("portArea")) { + return null; + } + String value = singleValue(params, "portArea"); + if (!StringUtils.hasText(value)) { + return 1; + } + String normalized = value.trim(); + if ("2".equals(normalized)) { + return 2; + } + if ("3".equals(normalized)) { + return 3; + } + throw new IllegalArgumentException("portArea must be empty, 2, or 3"); + } + + private String singleValue(MultiValueMap params, String name) { + List values = params.get(name); + if (values == null || values.size() != 1) { + throw new IllegalArgumentException(name + " must be supplied at most once"); + } + return values.get(0); + } +} diff --git a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationSubscription.java b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationSubscription.java new file mode 100644 index 0000000..e2858c0 --- /dev/null +++ b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationSubscription.java @@ -0,0 +1,27 @@ +package com.zcloud.personnel.positioning.websocket; + +final class PersonLocationSubscription { + static final String SESSION_ATTRIBUTE = PersonLocationSubscription.class.getName(); + + private final Long corpinfoId; + private final Integer portArea; + + PersonLocationSubscription(Long corpinfoId, Integer portArea) { + this.corpinfoId = corpinfoId; + this.portArea = portArea; + } + + boolean matches(Long messageCorpinfoId, Integer messagePortArea) { + boolean companyMatches = corpinfoId == null || corpinfoId.equals(messageCorpinfoId); + boolean portAreaMatches = portArea == null || portArea.equals(messagePortArea); + return companyMatches && portAreaMatches; + } + + Long getCorpinfoId() { + return corpinfoId; + } + + Integer getPortArea() { + return portArea; + } +} diff --git a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketConfiguration.java b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketConfiguration.java index e576e48..007c9b0 100644 --- a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketConfiguration.java +++ b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketConfiguration.java @@ -23,6 +23,7 @@ import java.util.List; ) public class PersonLocationWebSocketConfiguration implements WebSocketConfigurer { private final PersonLocationWebSocketHandler handler; + private final PersonLocationHandshakeInterceptor handshakeInterceptor; @Value("${application.gateway:personnelPosition}") private String gateway; @@ -33,6 +34,7 @@ public class PersonLocationWebSocketConfiguration implements WebSocketConfigurer @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(handler, endpointPath()) + .addInterceptors(handshakeInterceptor) .setAllowedOrigins(originValues()); } diff --git a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketHandler.java b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketHandler.java index 795e8d4..9936cec 100644 --- a/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketHandler.java +++ b/web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketHandler.java @@ -1,6 +1,7 @@ package com.zcloud.personnel.positioning.websocket; import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationPublisher; +import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; @@ -19,7 +20,7 @@ import java.util.concurrent.ConcurrentHashMap; public class PersonLocationWebSocketHandler extends TextWebSocketHandler implements RealtimeLocationPublisher { private static final Logger LOGGER = LoggerFactory.getLogger(PersonLocationWebSocketHandler.class); - private final Map sessions = new ConcurrentHashMap<>(); + private final Map sessions = new ConcurrentHashMap<>(); private final int sendTimeLimitMs; private final int bufferSizeLimitBytes; @@ -35,9 +36,11 @@ public class PersonLocationWebSocketHandler extends TextWebSocketHandler impleme public void afterConnectionEstablished(WebSocketSession session) { WebSocketSession concurrentSession = new ConcurrentWebSocketSessionDecorator( session, sendTimeLimitMs, bufferSizeLimitBytes); - sessions.put(session.getId(), concurrentSession); - LOGGER.info("Personnel location WebSocket connected, sessionId={}, connectionCount={}", - session.getId(), sessions.size()); + PersonLocationSubscription subscription = subscription(session); + sessions.put(session.getId(), new SubscriptionSession(concurrentSession, subscription)); + LOGGER.info("Personnel location WebSocket connected, sessionId={}, corpinfoId={}, portArea={}, " + + "connectionCount={}", + session.getId(), subscription.getCorpinfoId(), subscription.getPortArea(), sessions.size()); } @Override @@ -55,13 +58,18 @@ public class PersonLocationWebSocketHandler extends TextWebSocketHandler impleme } @Override - public int publish(String payload) { - TextMessage message = new TextMessage(payload); + public int publish(RealtimeLocationMessage locationMessage) { + TextMessage message = new TextMessage(locationMessage.getPayload()); int delivered = 0; - for (Map.Entry entry : sessions.entrySet()) { - WebSocketSession session = entry.getValue(); + for (Map.Entry entry : sessions.entrySet()) { + SubscriptionSession subscriptionSession = entry.getValue(); + if (!subscriptionSession.subscription.matches( + locationMessage.getCorpinfoId(), locationMessage.getPortArea())) { + continue; + } + WebSocketSession session = subscriptionSession.session; if (!session.isOpen()) { - sessions.remove(entry.getKey(), session); + sessions.remove(entry.getKey(), subscriptionSession); continue; } try { @@ -80,7 +88,8 @@ public class PersonLocationWebSocketHandler extends TextWebSocketHandler impleme } private void removeAndClose(String sessionId) { - WebSocketSession session = sessions.remove(sessionId); + SubscriptionSession subscriptionSession = sessions.remove(sessionId); + WebSocketSession session = subscriptionSession == null ? null : subscriptionSession.session; if (session == null || !session.isOpen()) { return; } @@ -90,4 +99,20 @@ public class PersonLocationWebSocketHandler extends TextWebSocketHandler impleme LOGGER.debug("Failed to close personnel location WebSocket, sessionId={}", sessionId, e); } } + + private PersonLocationSubscription subscription(WebSocketSession session) { + Object value = session.getAttributes().get(PersonLocationSubscription.SESSION_ATTRIBUTE); + return value instanceof PersonLocationSubscription + ? (PersonLocationSubscription) value : new PersonLocationSubscription(null, null); + } + + private static final class SubscriptionSession { + private final WebSocketSession session; + private final PersonLocationSubscription subscription; + + private SubscriptionSession(WebSocketSession session, PersonLocationSubscription subscription) { + this.session = session; + this.subscription = subscription; + } + } } diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/FindsStaffMappingSyncService.java b/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/FindsStaffMappingSyncService.java index 4834be4..47d5a04 100644 --- a/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/FindsStaffMappingSyncService.java +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/FindsStaffMappingSyncService.java @@ -4,6 +4,7 @@ import com.alibaba.cola.exception.BizException; import com.fasterxml.jackson.databind.JsonNode; import com.zcloud.personnel.positioning.command.FindsCorpMappingService; import com.zcloud.personnel.positioning.integration.finds.FindsOpenApiClient; +import com.zcloud.personnel.positioning.integration.realtime.RealtimePersonRoutingCache; import com.zcloud.personnel.positioning.persistence.repository.PersonnelMatchRepository; import lombok.RequiredArgsConstructor; import org.slf4j.Logger; @@ -39,6 +40,7 @@ public class FindsStaffMappingSyncService { private final PositionPersonLocalLookupService localLookupService; private final PersonnelMatchRepository personnelMatchRepository; private final FindsCorpMappingService findsCorpMappingService; + private final RealtimePersonRoutingCache realtimePersonRoutingCache; private FindsCorpMappingService.PathSyncSummary lastOrgPathSummary = FindsCorpMappingService.PathSyncSummary.empty(); @@ -68,6 +70,7 @@ public class FindsStaffMappingSyncService { } } disableMissingRows(syncTime); + realtimePersonRoutingCache.refreshNow(); LOGGER.info("FindS staff mapping synchronization completed, total={}, matched={}, " + "orgPaths={}, orgNodes={}, orgMatched={}, orgAmbiguous={}, orgUnmatched={}", staffRows.size(), matchedCount, orgSummary.getDistinctPathCount(), diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/RealtimePersonLocationMatcher.java b/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/RealtimePersonLocationMatcher.java index 7f6d85a..c97c69d 100644 --- a/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/RealtimePersonLocationMatcher.java +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/command/query/RealtimePersonLocationMatcher.java @@ -176,21 +176,23 @@ public class RealtimePersonLocationMatcher { "unifiedCreditCode", "creditNo"), directText(staff, "creditCode", "socialCreditCode", "unifiedSocialCreditCode", "unifiedCreditCode", "creditNo"))); - co.setPortArea(integerValue(firstValue(row, staff, "portArea", "portAreaCode"))); + // 本平台企业和港区信息在消息发布前统一从人员路由缓存补充。 + co.setPortArea(null); co.setTerminalNo(candidate.terminalNo); co.setPositionMode(positionMode(row)); - co.setLon(decimalValue(firstValue(location, row, "lon", "lng", "longitude"))); - co.setLat(decimalValue(firstValue(location, row, "lat", "latitude"))); - co.setAlt(decimalValue(firstValue(location, row, "alt", "altitude"))); + JsonNode position = firstObject(location, "position"); + co.setLon(decimalValue(firstValue(position, location, row, "lon", "lng", "longitude"))); + co.setLat(decimalValue(firstValue(position, location, row, "lat", "latitude"))); + co.setAlt(decimalValue(firstValue(position, location, row, "alt", "altitude"))); co.setSpeed(decimalValue(firstValue(location, row, "spd", "speed"))); co.setDirection(decimalValue(firstValue(location, row, "dir", "direction"))); co.setLastLocationTime(longValue(firstValue(location, row, "gt", "time", "locateTime", - "locationTime", "timestamp"))); + "locationTime", "timestamp", "epochTime", "serverTime"))); co.setCurrentLocation(firstText( directText(location, "fenceName", "areaName", "positionName", "location", "address"), directText(row, "fenceName", "areaName", "positionName", "locationName", "address"))); Boolean online = booleanValue(firstValue(location, row, "online", "isOnline")); - co.setOnline(online == null ? co.getLon() != null && co.getLat() != null : online); + co.setOnline(online == null ? Boolean.FALSE : online); co.setAlarmStatus(booleanValue(firstValue(location, row, "alarmStatus", "isAlarm"))); co.setAlarmCount(integerValue(firstValue(location, row, "alarmCount", "unclosedAlarmCount"))); co.setStaffRawJson(row.toString()); @@ -221,6 +223,14 @@ public class RealtimePersonLocationMatcher { return value == null ? directValue(second, fieldNames) : value; } + private JsonNode firstValue(JsonNode first, JsonNode second, JsonNode third, String... fieldNames) { + JsonNode value = directValue(first, fieldNames); + if (value == null) { + value = directValue(second, fieldNames); + } + return value == null ? directValue(third, fieldNames) : value; + } + private JsonNode directValue(JsonNode node, String... fieldNames) { if (node == null || !node.isObject()) { return null; diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java index 73e79f9..8e6cbfe 100644 --- a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java @@ -8,6 +8,9 @@ import com.zcloud.personnel.positioning.command.query.RealtimeLocationPayloadExc import com.zcloud.personnel.positioning.command.query.RealtimePersonLocationMatcher; import com.zcloud.personnel.positioning.dto.clientobject.BiPersonLocationCO; import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationPublisher; +import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationMessage; +import com.zcloud.personnel.positioning.integration.realtime.RealtimePersonRoutingCache; +import com.zcloud.personnel.positioning.integration.realtime.RealtimePersonRoutingCache.PersonRoute; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.header.Header; import org.slf4j.Logger; @@ -18,6 +21,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; import java.util.ArrayList; import java.util.Iterator; @@ -36,16 +40,19 @@ public class LocationKafkaProbeListener { private final int maxLogBytes; private final RealtimePersonLocationMatcher locationMatcher; + private final RealtimePersonRoutingCache personRoutingCache; private final ObjectMapper objectMapper; private final ObjectProvider publishers; public LocationKafkaProbeListener( @Value("${personnel-positioning.kafka.location.max-log-bytes:16384}") int maxLogBytes, RealtimePersonLocationMatcher locationMatcher, + RealtimePersonRoutingCache personRoutingCache, ObjectMapper objectMapper, ObjectProvider publishers) { this.maxLogBytes = Math.max(1, maxLogBytes); this.locationMatcher = locationMatcher; + this.personRoutingCache = personRoutingCache; this.objectMapper = objectMapper; this.publishers = publishers; } @@ -75,9 +82,12 @@ public class LocationKafkaProbeListener { List locations = locationMatcher.match(record.value()); int deliveries = 0; for (BiPersonLocationCO location : locations) { + enrichPersonRoute(location); String payload = serialize(location); + RealtimeLocationMessage message = new RealtimeLocationMessage( + payload, location.getCorpinfoId(), location.getPortArea()); for (RealtimeLocationPublisher publisher : publishers) { - deliveries += publisher.publish(payload); + deliveries += publisher.publish(message); } } acknowledgment.acknowledge(); @@ -91,6 +101,27 @@ public class LocationKafkaProbeListener { } } + private void enrichPersonRoute(BiPersonLocationCO location) { + if (location == null) { + return; + } + PersonRoute route = personRoutingCache.getRoute(location.getStaffNo()); + if (route == null) { + location.setCorpinfoId(null); + location.setPortArea(null); + return; + } + location.setUserId(route.getUserId()); + location.setLocalStaffNo(route.getLocalStaffNo()); + if (!StringUtils.hasText(location.getStaffName())) { + location.setStaffName(route.getStaffName()); + } + location.setCorpinfoId(route.getCorpinfoId()); + location.setCorpinfoName(route.getCorpinfoName()); + location.setCompanyName(route.getCorpinfoName()); + location.setPortArea(route.getPortArea()); + } + private String serialize(BiPersonLocationCO location) { try { ObjectNode payload = objectMapper.valueToTree(location); diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationMessage.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationMessage.java new file mode 100644 index 0000000..7929f4d --- /dev/null +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationMessage.java @@ -0,0 +1,25 @@ +package com.zcloud.personnel.positioning.integration.realtime; + +public final class RealtimeLocationMessage { + private final String payload; + private final Long corpinfoId; + private final Integer portArea; + + public RealtimeLocationMessage(String payload, Long corpinfoId, Integer portArea) { + this.payload = payload; + this.corpinfoId = corpinfoId; + this.portArea = portArea; + } + + public String getPayload() { + return payload; + } + + public Long getCorpinfoId() { + return corpinfoId; + } + + public Integer getPortArea() { + return portArea; + } +} diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationPublisher.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationPublisher.java index 7422d3a..61520ca 100644 --- a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationPublisher.java +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationPublisher.java @@ -1,5 +1,5 @@ package com.zcloud.personnel.positioning.integration.realtime; public interface RealtimeLocationPublisher { - int publish(String payload); + int publish(RealtimeLocationMessage message); } diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimePersonRoutingCache.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimePersonRoutingCache.java new file mode 100644 index 0000000..54f68e6 --- /dev/null +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimePersonRoutingCache.java @@ -0,0 +1,245 @@ +package com.zcloud.personnel.positioning.integration.realtime; + +import com.zcloud.personnel.positioning.persistence.dataobject.RealtimePersonRoutingDO; +import com.zcloud.personnel.positioning.persistence.repository.PersonnelMatchRepository; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * 每个 Pod 独立维护的人员路由快照，用于补全并筛选实时 WebSocket 消息。 + */ +@Component +public class RealtimePersonRoutingCache { + private static final Logger LOGGER = LoggerFactory.getLogger(RealtimePersonRoutingCache.class); + private static final String SOURCE_CONFIG_CODE = "FINDS"; + private static final long REFRESH_FAILURE_RETRY_MILLIS = 30_000L; + + private final PersonnelMatchRepository personnelMatchRepository; + private final long ttlMillis; + private final Object refreshMonitor = new Object(); + private volatile CacheSnapshot snapshot = CacheSnapshot.expired(); + + public RealtimePersonRoutingCache( + PersonnelMatchRepository personnelMatchRepository, + @Value("${personnel-positioning.kafka.location.person-routing-cache-ttl-ms:" + + "${personnel-positioning.kafka.location.staff-name-cache-ttl-ms:300000}}") long ttlMillis) { + this.personnelMatchRepository = personnelMatchRepository; + this.ttlMillis = Math.max(1_000L, ttlMillis); + } + + public PersonRoute getRoute(String staffNo) { + String key = normalize(staffNo); + return key == null ? null : currentSnapshot().routes.get(key); + } + + @EventListener(ApplicationReadyEvent.class) + public void warmUp() { + refresh(true); + } + + @Scheduled( + initialDelayString = "${personnel-positioning.kafka.location.person-routing-cache-check-ms:30000}", + fixedDelayString = "${personnel-positioning.kafka.location.person-routing-cache-check-ms:30000}" + ) + public void refreshIfExpired() { + refresh(false); + } + + public void refreshNow() { + refresh(true); + } + + private CacheSnapshot currentSnapshot() { + CacheSnapshot current = snapshot; + if (current.expireAt > System.currentTimeMillis()) { + return current; + } + return refresh(false); + } + + private CacheSnapshot refresh(boolean force) { + long now = System.currentTimeMillis(); + CacheSnapshot current = snapshot; + if (!force && current.expireAt > now) { + return current; + } + synchronized (refreshMonitor) { + current = snapshot; + now = System.currentTimeMillis(); + if (!force && current.expireAt > now) { + return current; + } + try { + Map routes = buildRoutes( + personnelMatchRepository.listRealtimePersonRoutes(SOURCE_CONFIG_CODE)); + snapshot = new CacheSnapshot(Collections.unmodifiableMap(routes), now + ttlMillis); + LOGGER.info("Realtime WebSocket person-routing cache refreshed, size={}", routes.size()); + } catch (RuntimeException e) { + snapshot = new CacheSnapshot( + current.routes, now + Math.min(ttlMillis, REFRESH_FAILURE_RETRY_MILLIS)); + LOGGER.warn("Refresh realtime WebSocket person-routing cache failed; retaining {} entries", + current.routes.size(), e); + } + return snapshot; + } + } + + private Map buildRoutes(List rows) { + Map persistedRoutes = new LinkedHashMap<>(); + UniqueRouteIndex usernames = new UniqueRouteIndex(); + UniqueRouteIndex phones = new UniqueRouteIndex(); + if (rows != null) { + for (RealtimePersonRoutingDO row : rows) { + if (row == null) { + continue; + } + PersonRoute route = toRoute(row); + if (row.isPersistedMatch()) { + String staffNo = normalize(row.getFindsStaffNo()); + if (staffNo != null) { + persistedRoutes.put(staffNo, route); + } + continue; + } + usernames.add(row.getLocalStaffNo(), route); + phones.add(row.getMobileNo(), route); + } + } + + Map routes = new LinkedHashMap<>(persistedRoutes); + Set fallbackKeys = new LinkedHashSet<>(); + fallbackKeys.addAll(usernames.keys()); + fallbackKeys.addAll(phones.keys()); + for (String key : fallbackKeys) { + if (routes.containsKey(key)) { + continue; + } + IndexResult username = usernames.resolve(key); + if (username.present) { + if (!username.ambiguous) { + routes.put(key, username.route); + } + continue; + } + IndexResult phone = phones.resolve(key); + if (phone.present && !phone.ambiguous) { + routes.put(key, phone.route); + } + } + return routes; + } + + private PersonRoute toRoute(RealtimePersonRoutingDO row) { + return new PersonRoute( + row.getUserId(), normalize(row.getLocalStaffNo()), normalize(row.getStaffName()), + row.getCorpinfoId(), normalize(row.getCorpinfoName()), row.getPortArea()); + } + + private static String normalize(String value) { + return StringUtils.hasText(value) ? value.trim() : null; + } + + public static final class PersonRoute { + private final Long userId; + private final String localStaffNo; + private final String staffName; + private final Long corpinfoId; + private final String corpinfoName; + private final Integer portArea; + + public PersonRoute(Long userId, String localStaffNo, String staffName, + Long corpinfoId, String corpinfoName, Integer portArea) { + this.userId = userId; + this.localStaffNo = localStaffNo; + this.staffName = staffName; + this.corpinfoId = corpinfoId; + this.corpinfoName = corpinfoName; + this.portArea = portArea; + } + + public Long getUserId() { return userId; } + public String getLocalStaffNo() { return localStaffNo; } + public String getStaffName() { return staffName; } + public Long getCorpinfoId() { return corpinfoId; } + public String getCorpinfoName() { return corpinfoName; } + public Integer getPortArea() { return portArea; } + } + + private static final class UniqueRouteIndex { + private final Map unique = new LinkedHashMap<>(); + private final Set ambiguous = new LinkedHashSet<>(); + + private void add(String rawKey, PersonRoute route) { + String key = normalize(rawKey); + if (key == null || ambiguous.contains(key)) { + return; + } + PersonRoute existing = unique.putIfAbsent(key, route); + if (existing != null && !sameUser(existing, route)) { + unique.remove(key); + ambiguous.add(key); + } + } + + private Set keys() { + Set keys = new LinkedHashSet<>(unique.keySet()); + keys.addAll(ambiguous); + return keys; + } + + private IndexResult resolve(String key) { + if (ambiguous.contains(key)) { + return IndexResult.ambiguous(); + } + PersonRoute route = unique.get(key); + return route == null ? IndexResult.absent() : IndexResult.unique(route); + } + + private boolean sameUser(PersonRoute first, PersonRoute second) { + return first.userId != null && first.userId.equals(second.userId); + } + } + + private static final class IndexResult { + private final boolean present; + private final boolean ambiguous; + private final PersonRoute route; + + private IndexResult(boolean present, boolean ambiguous, PersonRoute route) { + this.present = present; + this.ambiguous = ambiguous; + this.route = route; + } + + private static IndexResult absent() { return new IndexResult(false, false, null); } + private static IndexResult ambiguous() { return new IndexResult(true, true, null); } + private static IndexResult unique(PersonRoute route) { return new IndexResult(true, false, route); } + } + + private static final class CacheSnapshot { + private final Map routes; + private final long expireAt; + + private CacheSnapshot(Map routes, long expireAt) { + this.routes = routes; + this.expireAt = expireAt; + } + + private static CacheSnapshot expired() { + return new CacheSnapshot(Collections.emptyMap(), 0L); + } + } +} diff --git a/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeSchedulingConfiguration.java b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeSchedulingConfiguration.java new file mode 100644 index 0000000..37f74e7 --- /dev/null +++ b/web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeSchedulingConfiguration.java @@ -0,0 +1,9 @@ +package com.zcloud.personnel.positioning.integration.realtime; + +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableScheduling; + +@Configuration(proxyBeanMethods = false) +@EnableScheduling +public class RealtimeSchedulingConfiguration { +} diff --git a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/dataobject/RealtimePersonRoutingDO.java b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/dataobject/RealtimePersonRoutingDO.java new file mode 100644 index 0000000..6c6ad1a --- /dev/null +++ b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/dataobject/RealtimePersonRoutingDO.java @@ -0,0 +1,9 @@ +package com.zcloud.personnel.positioning.persistence.dataobject; + +import lombok.Data; + +@Data +public class RealtimePersonRoutingDO extends PersonnelLocalPersonDO { + private String findsStaffNo; + private boolean persistedMatch; +} diff --git a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/mapper/PersonnelMatchMapper.java b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/mapper/PersonnelMatchMapper.java index 3267386..0bcace1 100644 --- a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/mapper/PersonnelMatchMapper.java +++ b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/mapper/PersonnelMatchMapper.java @@ -2,6 +2,7 @@ package com.zcloud.personnel.positioning.persistence.mapper; import com.zcloud.personnel.positioning.persistence.dataobject.PersonnelLocalPersonDO; import com.zcloud.personnel.positioning.persistence.dataobject.PersonnelMappingSnapshotDO; +import com.zcloud.personnel.positioning.persistence.dataobject.RealtimePersonRoutingDO; import com.zcloud.personnel.positioning.persistence.dataobject.TerminalBindingSnapshotDO; import com.zcloud.personnel.positioning.persistence.dataobject.TerminalLocationSnapshotDO; import com.zcloud.personnel.positioning.persistence.dataobject.CorpInfoSnapshotDO; @@ -45,6 +46,9 @@ public interface PersonnelMatchMapper { int markStaleOffline(@Param("sourceConfigCode") String sourceConfigCode, @Param("syncTime") LocalDateTime syncTime); + List listRealtimePersonRoutes( + @Param("sourceConfigCode") String sourceConfigCode); + List listMatched(@Param("params") Map params); List listPeopleByPhones(@Param("values") List values); diff --git a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/PersonnelMatchRepository.java b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/PersonnelMatchRepository.java index f439042..52a10e4 100644 --- a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/PersonnelMatchRepository.java +++ b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/PersonnelMatchRepository.java @@ -2,6 +2,7 @@ package com.zcloud.personnel.positioning.persistence.repository; import com.zcloud.personnel.positioning.persistence.dataobject.PersonnelLocalPersonDO; import com.zcloud.personnel.positioning.persistence.dataobject.PersonnelMappingSnapshotDO; +import com.zcloud.personnel.positioning.persistence.dataobject.RealtimePersonRoutingDO; import com.zcloud.personnel.positioning.persistence.dataobject.TerminalBindingSnapshotDO; import com.zcloud.personnel.positioning.persistence.dataobject.TerminalLocationSnapshotDO; import com.zcloud.personnel.positioning.persistence.dataobject.CorpInfoSnapshotDO; @@ -30,6 +31,8 @@ public interface PersonnelMatchRepository { void markStaleOffline(String sourceConfigCode, LocalDateTime syncTime); + List listRealtimePersonRoutes(String sourceConfigCode); + List listMatched(Map params); List listPeopleByPhones(List values); diff --git a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/impl/PersonnelMatchRepositoryImpl.java b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/impl/PersonnelMatchRepositoryImpl.java index aece3ec..2228cc0 100644 --- a/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/impl/PersonnelMatchRepositoryImpl.java +++ b/web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/impl/PersonnelMatchRepositoryImpl.java @@ -2,6 +2,7 @@ package com.zcloud.personnel.positioning.persistence.repository.impl; import com.zcloud.personnel.positioning.persistence.dataobject.PersonnelLocalPersonDO; import com.zcloud.personnel.positioning.persistence.dataobject.PersonnelMappingSnapshotDO; +import com.zcloud.personnel.positioning.persistence.dataobject.RealtimePersonRoutingDO; import com.zcloud.personnel.positioning.persistence.dataobject.TerminalBindingSnapshotDO; import com.zcloud.personnel.positioning.persistence.dataobject.TerminalLocationSnapshotDO; import com.zcloud.personnel.positioning.persistence.dataobject.CorpInfoSnapshotDO; @@ -42,6 +43,9 @@ public class PersonnelMatchRepositoryImpl implements PersonnelMatchRepository { @Override public void markStaleOffline(String code, LocalDateTime time) { personnelMatchMapper.markStaleOffline(code, time); } + @Override public List listRealtimePersonRoutes(String code) { + return personnelMatchMapper.listRealtimePersonRoutes(code); + } @Override public List listMatched(Map params) { return personnelMatchMapper.listMatched(params); } diff --git a/web-infrastructure/src/main/resources/mapper/PersonnelMatchMapper.xml b/web-infrastructure/src/main/resources/mapper/PersonnelMatchMapper.xml index 867182a..7005265 100644 --- a/web-infrastructure/src/main/resources/mapper/PersonnelMatchMapper.xml +++ b/web-infrastructure/src/main/resources/mapper/PersonnelMatchMapper.xml @@ -85,6 +85,38 @@ AND (location_status_time IS NULL OR location_status_time < #{syncTime}) + + SELECT u.id AS user_id, u.username AS local_staff_no, u.name AS staff_name, TRIM(u.phone) AS mobile_no, u.user_id_card AS id_card_no, ci.id AS corpinfo_id,