zcloud_gbs
/
zcloud_gbs_personnel_positioning
14
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 支持人员定位WebSocket企业与港区过滤

master
shenzhidan 2026-08-12 15:04:13 +08:00
parent 1f585a01ef
commit 4d24075d22
18 changed files with 549 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
start/src/main/resources/kafka.yml
View File

@ -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}

5
web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/MockPersonLocationPublisher.java
View File

@ -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);

95
web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationHandshakeInterceptor.java Normal file
View File

@ -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<String, Object> attributes) {
try {
MultiValueMap<String, String> 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<String, String> 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<String, String> 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<String, String> params, String name) {
List<String> values = params.get(name);
if (values == null || values.size() != 1) {
throw new IllegalArgumentException(name + " must be supplied at most once");
}
return values.get(0);
}
}

27
web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationSubscription.java Normal file
View File

@ -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;
}
}

2
web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketConfiguration.java
View File

@ -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());
}

45
web-adapter/src/main/java/com/zcloud/personnel/positioning/websocket/PersonLocationWebSocketHandler.java
View File

@ -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<String, WebSocketSession> sessions = new ConcurrentHashMap<>();
private final Map<String, SubscriptionSession> 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<String, WebSocketSession> entry : sessions.entrySet()) {
WebSocketSession session = entry.getValue();
for (Map.Entry<String, SubscriptionSession> 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;
}
}
}

3
web-app/src/main/java/com/zcloud/personnel/positioning/command/query/FindsStaffMappingSyncService.java
View File

@ -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(),

22
web-app/src/main/java/com/zcloud/personnel/positioning/command/query/RealtimePersonLocationMatcher.java
View File

@ -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;

33
web-app/src/main/java/com/zcloud/personnel/positioning/integration/kafka/LocationKafkaProbeListener.java
View File

@ -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<RealtimeLocationPublisher> publishers;
public LocationKafkaProbeListener(
@Value("${personnel-positioning.kafka.location.max-log-bytes:16384}") int maxLogBytes,
RealtimePersonLocationMatcher locationMatcher,
RealtimePersonRoutingCache personRoutingCache,
ObjectMapper objectMapper,
ObjectProvider<RealtimeLocationPublisher> 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<BiPersonLocationCO> 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);

25
web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationMessage.java Normal file
View File

@ -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;
}
}

2
web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeLocationPublisher.java
View File

@ -1,5 +1,5 @@
package com.zcloud.personnel.positioning.integration.realtime;
public interface RealtimeLocationPublisher {
int publish(String payload);
int publish(RealtimeLocationMessage message);
}

245
web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimePersonRoutingCache.java Normal file
View File

@ -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<String, PersonRoute> 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<String, PersonRoute> buildRoutes(List<RealtimePersonRoutingDO> rows) {
Map<String, PersonRoute> 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<String, PersonRoute> routes = new LinkedHashMap<>(persistedRoutes);
Set<String> 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<String, PersonRoute> unique = new LinkedHashMap<>();
private final Set<String> 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<String> keys() {
Set<String> 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<String, PersonRoute> routes;
private final long expireAt;
private CacheSnapshot(Map<String, PersonRoute> routes, long expireAt) {
this.routes = routes;
this.expireAt = expireAt;
}
private static CacheSnapshot expired() {
return new CacheSnapshot(Collections.emptyMap(), 0L);
}
}
}

9
web-app/src/main/java/com/zcloud/personnel/positioning/integration/realtime/RealtimeSchedulingConfiguration.java Normal file
View File

@ -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 {
}

9
web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/dataobject/RealtimePersonRoutingDO.java Normal file
View File

@ -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;
}

4
web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/mapper/PersonnelMatchMapper.java
View File

@ -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<RealtimePersonRoutingDO> listRealtimePersonRoutes(
@Param("sourceConfigCode") String sourceConfigCode);
List<PersonnelMappingSnapshotDO> listMatched(@Param("params") Map<String, Object> params);
List<PersonnelLocalPersonDO> listPeopleByPhones(@Param("values") List<String> values);

3
web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/PersonnelMatchRepository.java
View File

@ -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<RealtimePersonRoutingDO> listRealtimePersonRoutes(String sourceConfigCode);
List<PersonnelMappingSnapshotDO> listMatched(Map<String, Object> params);
List<PersonnelLocalPersonDO> listPeopleByPhones(List<String> values);

4
web-infrastructure/src/main/java/com/zcloud/personnel/positioning/persistence/repository/impl/PersonnelMatchRepositoryImpl.java
View File

@ -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<RealtimePersonRoutingDO> listRealtimePersonRoutes(String code) {
return personnelMatchMapper.listRealtimePersonRoutes(code);
}
@Override public List<PersonnelMappingSnapshotDO> listMatched(Map<String, Object> params) {
return personnelMatchMapper.listMatched(params);
}

32
web-infrastructure/src/main/resources/mapper/PersonnelMatchMapper.xml
View File

@ -85,6 +85,38 @@
AND (location_status_time IS NULL OR location_status_time &lt; #{syncTime})
</update>
<select id="listRealtimePersonRoutes"
resultType="com.zcloud.personnel.positioning.persistence.dataobject.RealtimePersonRoutingDO">
SELECT pm.finds_staff_no, 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, ci.corp_name AS corpinfo_name, ci.code AS credit_code,
ci.port_area AS port_area, u.department_id, d.dept_name AS department_name,
1 AS persisted_match
FROM personnel_match pm
INNER JOIN user_scope_v u ON u.id = pm.local_user_id
LEFT JOIN corp_info ci ON ci.id = u.corpinfo_id
AND (ci.delete_enum IS NULL OR ci.delete_enum = 'FALSE')
LEFT JOIN department d ON d.id = u.department_id
AND (d.delete_enum IS NULL OR d.delete_enum = 'FALSE')
WHERE pm.source_config_code = #{sourceConfigCode}
AND pm.match_status = 'MATCHED'
AND (pm.delete_enum IS NULL OR pm.delete_enum = 'FALSE')
AND (u.delete_enum IS NULL OR u.delete_enum = 'FALSE')
AND pm.finds_staff_no IS NOT NULL AND TRIM(pm.finds_staff_no) &lt;&gt; ''
UNION ALL
SELECT NULL AS finds_staff_no, 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, ci.corp_name AS corpinfo_name, ci.code AS credit_code,
ci.port_area AS port_area, u.department_id, d.dept_name AS department_name,
0 AS persisted_match
FROM user_scope_v u
LEFT JOIN corp_info ci ON ci.id = u.corpinfo_id
AND (ci.delete_enum IS NULL OR ci.delete_enum = 'FALSE')
LEFT JOIN department d ON d.id = u.department_id
AND (d.delete_enum IS NULL OR d.delete_enum = 'FALSE')
WHERE (u.delete_enum IS NULL OR u.delete_enum = 'FALSE')
</select>
<sql id="localPersonSelect">
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,