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

优化人员同步与实时定位处理

master
shenzhidan 2026-08-13 18:23:45 +08:00
parent de78e0c26a
commit 8bbe0e7932
5 changed files with 160 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -11,11 +11,12 @@ import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* Refreshes the persisted online/offline state for FindS personnel.
@ -33,7 +34,6 @@ public class FindsStaffLocationStatusSyncService {
private static final int STAFF_COUNT_PAGE_SIZE = 1;
private static final int TERMINAL_PAGE_SIZE = 2000;
private static final int MAX_TERMINAL_PAGE_COUNT = 100;
private static final int BATCH_CONCURRENCY = 4;
private static final int FULL_SYNC_INTERVAL_HOURS = 24;
private static final List<String> ONLINE_DEVICE_STATUSES =
Arrays.asList("STATIC", "MOVING");
@ -51,15 +51,14 @@ public class FindsStaffLocationStatusSyncService {
|| fullSyncDue(syncTime);
int fullSyncCount = fullSync ? mappingSyncService.syncAll() : 0;
Map<String, OnlineTerminal> onlineTerminals = loadOnlineTerminals();
int onlineMappings = updateLocationStatuses(onlineTerminals, syncTime);
int onlineStaffTerminals = countStaffTerminals(onlineTerminals);
StatusSyncCounter statusCounter = synchronizeOnlineTerminals(syncTime);
personnelMatchRepository.markStaleOffline(SOURCE_CONFIG_CODE, syncTime);
LOGGER.info("FindS personnel location status synchronization completed, remoteStaffCount={}, "
+ "localStaffCount={}, fullSync={}, fullSyncCount={}, onlineTerminals={}, onlineMappings={}",
remoteStaffCount, localStaffCount, fullSync, fullSyncCount,
onlineStaffTerminals, onlineMappings);
statusCounter.onlineStaffTerminals, statusCounter.onlineMappings);
return new SyncSummary(remoteStaffCount, localStaffCount, fullSync, fullSyncCount,
onlineStaffTerminals, onlineMappings);
statusCounter.onlineStaffTerminals, statusCounter.onlineMappings);
}
private long queryRemoteStaffCount() {
@ -85,15 +84,17 @@ public class FindsStaffLocationStatusSyncService {
|| lastSyncTime.isBefore(syncTime.minusHours(FULL_SYNC_INTERVAL_HOURS));
}
private Map<String, OnlineTerminal> loadOnlineTerminals() {
Map<String, OnlineTerminal> result = new LinkedHashMap<>();
private StatusSyncCounter synchronizeOnlineTerminals(LocalDateTime syncTime) {
StatusSyncCounter counter = new StatusSyncCounter();
for (String status : ONLINE_DEVICE_STATUSES) {
loadOnlineTerminals(status, result);
synchronizeOnlineTerminals(status, syncTime, counter);
}
return result;
return counter;
}
private void loadOnlineTerminals(String deviceStatus, Map<String, OnlineTerminal> result) {
private void synchronizeOnlineTerminals(String deviceStatus,
LocalDateTime syncTime,
StatusSyncCounter counter) {
JsonNode firstData = successData(findsOpenApiClient.postFailFast(
TERMINAL_LIST_API, terminalPageRequest(deviceStatus, 1)));
JsonNode firstRows = extractRows(firstData);
@ -101,7 +102,7 @@ public class FindsStaffLocationStatusSyncService {
if (total > 0 && (firstRows == null || firstRows.size() == 0)) {
throw new BizException("FindS terminal status page is empty while total is " + total);
}
appendOnlineTerminals(result, firstRows, deviceStatus);
synchronizeTerminalPage(firstRows, deviceStatus, syncTime, counter);
int effectivePageSize = firstRows == null || firstRows.size() == 0
? TERMINAL_PAGE_SIZE : firstRows.size();
@ -109,20 +110,13 @@ public class FindsStaffLocationStatusSyncService {
if (pageCount > MAX_TERMINAL_PAGE_COUNT) {
throw new BizException("FindS online terminal pages exceed the configured safety limit: " + pageCount);
}
if (pageCount <= 1) {
return;
}
List<Map<String, Object>> requests = new ArrayList<>(pageCount - 1);
for (int pageNo = 2; pageNo <= pageCount; pageNo++) {
requests.add(terminalPageRequest(deviceStatus, pageNo));
}
long receivedCount = firstRows == null ? 0 : firstRows.size();
for (JsonNode root : findsOpenApiClient.postFailFastBatch(
TERMINAL_LIST_API, requests, BATCH_CONCURRENCY)) {
JsonNode rows = extractRows(successData(root));
for (int pageNo = 2; pageNo <= pageCount; pageNo++) {
JsonNode data = successData(findsOpenApiClient.postFailFast(
TERMINAL_LIST_API, terminalPageRequest(deviceStatus, pageNo)));
JsonNode rows = extractRows(data);
receivedCount += rows == null ? 0 : rows.size();
appendOnlineTerminals(result, rows, deviceStatus);
synchronizeTerminalPage(rows, deviceStatus, syncTime, counter);
}
if (receivedCount < total) {
throw new BizException("FindS online terminal result is incomplete: received="
@ -138,9 +132,10 @@ public class FindsStaffLocationStatusSyncService {
return request;
}
private void appendOnlineTerminals(Map<String, OnlineTerminal> result,
JsonNode rows,
String requestedStatus) {
private void synchronizeTerminalPage(JsonNode rows,
String requestedStatus,
LocalDateTime syncTime,
StatusSyncCounter counter) {
if (rows == null || !rows.isArray()) {
return;
}
@ -150,6 +145,9 @@ public class FindsStaffLocationStatusSyncService {
if (!StringUtils.hasText(terminalNo)) {
continue;
}
if (!counter.seenTerminalNos.add(terminalNo)) {
continue;
}
String trackType = firstText(
text(firstPresent(row, "trackType")),
text(firstPresent(row.path("track"), "trackType")));
@ -159,35 +157,16 @@ public class FindsStaffLocationStatusSyncService {
boolean staffTrack = "STAFF".equalsIgnoreCase(trackType) && StringUtils.hasText(trackNo);
String actualStatus = firstText(
text(firstPresent(row, "deviceStatus", "status", "onlineStatus")), requestedStatus);
result.put(terminalNo, new OnlineTerminal(
terminalNo, actualStatus, staffTrack ? trackNo.trim() : null));
}
}
private int updateLocationStatuses(Map<String, OnlineTerminal> onlineTerminals,
LocalDateTime syncTime) {
int matched = 0;
for (OnlineTerminal terminal : onlineTerminals.values()) {
String staffNo = staffTrack ? trackNo.trim() : null;
personnelMatchRepository.refreshTerminalBinding(
SOURCE_CONFIG_CODE, terminal.terminalNo, terminal.staffNo, syncTime);
if (!StringUtils.hasText(terminal.staffNo)) {
SOURCE_CONFIG_CODE, terminalNo, staffNo, syncTime);
if (!StringUtils.hasText(staffNo)) {
continue;
}
matched += personnelMatchRepository.updateOnline(
SOURCE_CONFIG_CODE, terminal.terminalNo, terminal.deviceStatus, syncTime);
counter.onlineStaffTerminals++;
counter.onlineMappings += personnelMatchRepository.updateOnline(
SOURCE_CONFIG_CODE, terminalNo, actualStatus, syncTime);
}
personnelMatchRepository.markStaleOffline(SOURCE_CONFIG_CODE, syncTime);
return matched;
}
private int countStaffTerminals(Map<String, OnlineTerminal> onlineTerminals) {
int count = 0;
for (OnlineTerminal terminal : onlineTerminals.values()) {
if (StringUtils.hasText(terminal.staffNo)) {
count++;
}
}
return count;
}
private JsonNode successData(JsonNode root) {
@ -260,16 +239,10 @@ public class FindsStaffLocationStatusSyncService {
}
}
private static class OnlineTerminal {
private final String terminalNo;
private final String deviceStatus;
private final String staffNo;
private OnlineTerminal(String terminalNo, String deviceStatus, String staffNo) {
this.terminalNo = terminalNo;
this.deviceStatus = deviceStatus;
this.staffNo = staffNo;
}
private static class StatusSyncCounter {
private final Set<String> seenTerminalNos = new HashSet<>();
private int onlineStaffTerminals;
private int onlineMappings;
}
public static class SyncSummary {

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

@ -34,7 +34,6 @@ public class FindsStaffMappingSyncService {
private static final String SOURCE_CONFIG_CODE = "FINDS";
private static final int PAGE_SIZE = 2000;
private static final int MAX_PAGE_COUNT = 50;
private static final int BATCH_CONCURRENCY = 4;
private final FindsOpenApiClient findsOpenApiClient;
private final PositionPersonLocalLookupService localLookupService;
@ -51,62 +50,62 @@ public class FindsStaffMappingSyncService {
/** 拉取 FindS 完整组织路径并单独刷新企业映射，不执行人员映射落库。 */
public FindsCorpMappingService.PathSyncSummary syncOrganizations() {
LocalDateTime syncTime = LocalDateTime.now().withNano(0);
return synchronizeOrganizations(loadAllStaff(), syncTime);
Set<List<String>> orgPaths = collectOrganizationPaths();
FindsCorpMappingService.PathSyncSummary summary = synchronizeOrganizationPaths(orgPaths, syncTime);
if (orgPaths.isEmpty()) {
forEachStaffPage(rows -> resolveOrganizations(rows));
}
return summary;
}
public int syncAll() {
// The schema uses second precision; keep the same precision for the stale-row cutoff.
// 数据库时间精度为秒，保持同一精度才能准确识别本次未出现的旧数据。
LocalDateTime syncTime = LocalDateTime.now().withNano(0);
List<StaffSnapshot> staffRows = loadAllStaff();
FindsCorpMappingService.PathSyncSummary orgSummary =
synchronizeOrganizations(staffRows, syncTime);
Set<List<String>> orgPaths = collectOrganizationPaths();
FindsCorpMappingService.PathSyncSummary orgSummary = synchronizeOrganizationPaths(orgPaths, syncTime);
PersonIndexes localPeople = new PersonIndexes(localLookupService.findAllPeople());
int matchedCount = 0;
for (StaffSnapshot row : staffRows) {
LocalMatch localMatch = localPeople.match(row);
upsert(row, localMatch, syncTime);
if (localMatch.person != null) {
matchedCount++;
}
}
SyncCounter counter = new SyncCounter();
forEachStaffPage(rows -> syncStaffPage(
rows, localPeople, syncTime, orgPaths.isEmpty(), counter));
disableMissingRows(syncTime);
realtimePersonRoutingCache.refreshNow();
LOGGER.info("FindS staff mapping synchronization completed, total={}, matched={}, "
+ "orgPaths={}, orgNodes={}, orgMatched={}, orgAmbiguous={}, orgUnmatched={}",
staffRows.size(), matchedCount, orgSummary.getDistinctPathCount(),
counter.total, counter.matched, orgSummary.getDistinctPathCount(),
orgSummary.getDistinctNodeCount(), orgSummary.getMatchedCount(),
orgSummary.getAmbiguousCount(), orgSummary.getUnmatchedCount());
return staffRows.size();
return counter.total;
}
private FindsCorpMappingService.PathSyncSummary synchronizeOrganizations(
List<StaffSnapshot> staffRows, LocalDateTime syncTime) {
private Set<List<String>> collectOrganizationPaths() {
Set<List<String>> orgPaths = new LinkedHashSet<>();
for (StaffSnapshot row : staffRows) {
if (row.orgNamePath != null && !row.orgNamePath.isEmpty()) {
orgPaths.add(row.orgNamePath);
forEachStaffPage(rows -> {
for (JsonNode row : rows) {
List<String> path = orgNamePath(row.path("orgNamePath"));
if (!path.isEmpty()) {
orgPaths.add(path);
}
}
}
});
return orgPaths;
}
private FindsCorpMappingService.PathSyncSummary synchronizeOrganizationPaths(
Set<List<String>> orgPaths, LocalDateTime syncTime) {
FindsCorpMappingService.PathSyncSummary orgSummary =
findsCorpMappingService.synchronizePaths(orgPaths, syncTime);
if (orgSummary == null) {
orgSummary = FindsCorpMappingService.PathSyncSummary.empty();
}
lastOrgPathSummary = orgSummary;
if (orgPaths.isEmpty()) {
for (StaffSnapshot row : staffRows) {
findsCorpMappingService.resolve(SOURCE_CONFIG_CODE, row.findsCorpinfoId,
row.corpinfoName, row.departmentName, null);
}
}
return orgSummary;
}
private List<StaffSnapshot> loadAllStaff() {
List<StaffSnapshot> result = new ArrayList<>();
private void forEachStaffPage(StaffPageConsumer consumer) {
JsonNode firstData = successData(
findsOpenApiClient.postFailFast(STAFF_LIST_API, pageRequest(1)));
int firstPageSize = appendRows(result, extractRows(firstData));
JsonNode firstRows = extractRows(firstData);
int firstPageSize = rowCount(firstRows);
long total = requiredTotal(firstData);
if (total > 0 && firstPageSize == 0) {
throw new BizException("FindS staff page is empty while total is " + total);
@ -115,23 +114,19 @@ public class FindsStaffMappingSyncService {
if (pageCount > MAX_PAGE_COUNT) {
throw new BizException("FindS staff pages exceed the configured safety limit: " + pageCount);
}
if (pageCount <= 1) {
return result;
}
List<Map<String, Object>> requests = new ArrayList<>(pageCount - 1);
for (int pageNo = 2; pageNo <= pageCount; pageNo++) {
requests.add(pageRequest(pageNo));
}
consumer.accept(firstRows);
long receivedCount = firstPageSize;
for (JsonNode root : findsOpenApiClient.postFailFastBatch(
STAFF_LIST_API, requests, BATCH_CONCURRENCY)) {
receivedCount += appendRows(result, extractRows(successData(root)));
for (int pageNo = 2; pageNo <= pageCount; pageNo++) {
JsonNode data = successData(
findsOpenApiClient.postFailFast(STAFF_LIST_API, pageRequest(pageNo)));
JsonNode rows = extractRows(data);
receivedCount += rowCount(rows);
consumer.accept(rows);
}
if (receivedCount < total) {
throw new BizException("FindS staff result is incomplete: received="
+ receivedCount + ", total=" + total);
}
return result;
}
private Map<String, Object> pageRequest(int pageNo) {
@ -141,17 +136,47 @@ public class FindsStaffMappingSyncService {
return request;
}
private int appendRows(List<StaffSnapshot> result, JsonNode rows) {
private int rowCount(JsonNode rows) {
return rows != null && rows.isArray() ? rows.size() : 0;
}
private void resolveOrganizations(JsonNode rows) {
if (rows == null || !rows.isArray()) {
return 0;
return;
}
for (JsonNode row : rows) {
StaffSnapshot snapshot = map(row);
if (snapshot != null) {
result.add(snapshot);
findsCorpMappingService.resolve(SOURCE_CONFIG_CODE, snapshot.findsCorpinfoId,
snapshot.corpinfoName, snapshot.departmentName, null);
}
}
}
private void syncStaffPage(JsonNode rows,
PersonIndexes localPeople,
LocalDateTime syncTime,
boolean resolveOrganization,
SyncCounter counter) {
if (rows == null || !rows.isArray()) {
return;
}
for (JsonNode row : rows) {
StaffSnapshot snapshot = map(row);
if (snapshot == null) {
continue;
}
if (resolveOrganization) {
findsCorpMappingService.resolve(SOURCE_CONFIG_CODE, snapshot.findsCorpinfoId,
snapshot.corpinfoName, snapshot.departmentName, null);
}
LocalMatch localMatch = localPeople.match(snapshot);
upsert(snapshot, localMatch, syncTime);
counter.total++;
if (localMatch.person != null) {
counter.matched++;
}
}
return rows.size();
}
private StaffSnapshot map(JsonNode row) {
@ -355,6 +380,16 @@ public class FindsStaffMappingSyncService {
private String terminalNo;
}
@FunctionalInterface
private interface StaffPageConsumer {
void accept(JsonNode rows);
}
private static class SyncCounter {
private int total;
private int matched;
}
private static class PersonIndexes {
private final UniqueIndex idCards = new UniqueIndex();
private final UniqueIndex usernames = new UniqueIndex();

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

@ -12,11 +12,9 @@ import com.zcloud.personnel.positioning.integration.realtime.RealtimeLocationMes
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;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
@ -27,7 +25,6 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.StringJoiner;
@Component
@ConditionalOnProperty(
@ -38,19 +35,16 @@ import java.util.StringJoiner;
public class LocationKafkaProbeListener {
private static final Logger LOGGER = LoggerFactory.getLogger(LocationKafkaProbeListener.class);
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;
@ -63,21 +57,6 @@ public class LocationKafkaProbeListener {
groupId = "${personnel-positioning.kafka.location.group-id:personnel-position-location-probe}"
)
public void onMessage(ConsumerRecord<byte[], byte[]> record, Acknowledgment acknowledgment) {
LOGGER.info(
"收到Kafka定位原始消息，topic={}, partition={}, offset={}, timestamp={}, "
+ "timestampType={}, keyBytes={}, valueBytes={}, key={}, headers={}, value={}",
record.topic(),
record.partition(),
record.offset(),
record.timestamp(),
record.timestampType(),
length(record.key()),
length(record.value()),
LocationKafkaPayloadPreview.render(record.key(), maxLogBytes),
renderHeaders(record),
LocationKafkaPayloadPreview.render(record.value(), maxLogBytes)
);
try {
List<BiPersonLocationCO> locations = locationMatcher.match(record.value());
int deliveries = 0;
@ -91,9 +70,6 @@ public class LocationKafkaProbeListener {
}
}
acknowledgment.acknowledge();
LOGGER.info("Kafka location message processed, topic={}, partition={}, offset={}, converted={}, "
+ "websocketDeliveries={}",
record.topic(), record.partition(), record.offset(), locations.size(), deliveries);
} catch (RealtimeLocationPayloadException e) {
acknowledgment.acknowledge();
LOGGER.warn("Ignored invalid Kafka location message, topic={}, partition={}, offset={}, reason={}",
@ -140,15 +116,4 @@ public class LocationKafkaProbeListener {
}
}
private int length(byte[] value) {
return value == null ? 0 : value.length;
}
private String renderHeaders(ConsumerRecord<byte[], byte[]> record) {
StringJoiner joiner = new StringJoiner(", ", "[", "]");
for (Header header : record.headers()) {
joiner.add(header.key() + "=" + LocationKafkaPayloadPreview.render(header.value(), maxLogBytes));
}
return joiner.toString();
}
}

40
web-infrastructure/src/main/resources/db/migration/mysql/V20260811_01__personnel_location_area.sql Normal file
View File

@ -0,0 +1,40 @@
-- 保存实时定位点匹配出的当前区域；不保存实时经纬度和轨迹点。
SET @ddl = IF(
(SELECT COUNT(1) FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'personnel_match' AND COLUMN_NAME = 'location_area_id') = 0,
'ALTER TABLE `personnel_match` ADD COLUMN `location_area_id` bigint DEFAULT NULL COMMENT ''当前定位区域id'' AFTER `location_status_time`',
'DO 0'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @ddl = IF(
(SELECT COUNT(1) FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'personnel_match' AND COLUMN_NAME = 'location_area_name') = 0,
'ALTER TABLE `personnel_match` ADD COLUMN `location_area_name` varchar(255) DEFAULT NULL COMMENT ''当前定位区域名称'' AFTER `location_area_id`',
'DO 0'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @ddl = IF(
(SELECT COUNT(1) FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'personnel_match' AND COLUMN_NAME = 'location_area_time') = 0,
'ALTER TABLE `personnel_match` ADD COLUMN `location_area_time` datetime DEFAULT NULL COMMENT ''定位区域状态更新时间'' AFTER `location_area_name`',
'DO 0'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @ddl = IF(
(SELECT COUNT(1) FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'personnel_match' AND INDEX_NAME = 'idx_personnel_match_location_area') = 0,
'ALTER TABLE `personnel_match` ADD KEY `idx_personnel_match_location_area` (`location_area_id`, `location_status`)',
'DO 0'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

3
web-infrastructure/src/main/resources/db/migration/mysql/V20260812_01__expand_org_match_type.sql Normal file
View File

@ -0,0 +1,3 @@
-- 组织名称标准化匹配类型超过 32 个字符，扩容以保证人员全量同步可以继续执行。
ALTER TABLE `org_match`
MODIFY COLUMN `match_type` varchar(64) DEFAULT NULL COMMENT '组织匹配方式';