package com.xiaopeng.admin.scheduler;
|
|
import com.corundumstudio.socketio.BroadcastOperations;
|
import com.corundumstudio.socketio.SocketIOClient;
|
import com.corundumstudio.socketio.SocketIOServer;
|
import com.xiaopeng.admin.service.CommandCenterService;
|
import com.xiaopeng.admin.service.CommandCenterService.AggregateResult;
|
import com.xiaopeng.admin.service.PassengerCallService;
|
import com.xiaopeng.admin.socketio.SocketIOSessionManager;
|
import com.xiaopeng.admin.support.CommandCenterMockDetailSupport;
|
import com.xiaopeng.admin.support.CommandCenterMockKpiSupport;
|
import com.xiaopeng.api.vo.commandCenter.CommandCenterKpiVo;
|
import com.xiaopeng.api.vo.commandCenter.CommandCenterVehicleDetailVo;
|
import com.xiaopeng.api.vo.commandCenter.PassengerCallVo;
|
import com.xiaopeng.api.vo.commandCenter.VehicleMonitorVo;
|
import com.xiaopeng.api.dto.WeatherDTO;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.cloud.context.config.annotation.RefreshScope;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.Resource;
|
import java.time.LocalDateTime;
|
import java.util.ArrayList;
|
import java.util.Arrays;
|
import java.util.Collections;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
/**
|
* 指挥中心定时推送
|
*
|
* <p>按概设要求拆分推送频率：</p>
|
* <ul>
|
* <li>车辆位置/状态：2 秒一次（pushVehicles）</li>
|
* <li>KPI / 呼单 / 天气：5 秒一次（pushKpiAndCalls）</li>
|
* <li>单车详情：2 秒一次，仅推送给已订阅详情的客户端（pushVehicleDetails）</li>
|
* </ul>
|
*
|
* @ClassName CommandCenterPushScheduler
|
*/
|
@Slf4j
|
@Component
|
@RefreshScope
|
public class CommandCenterPushScheduler {
|
|
private static final String NAMESPACE = "/commandCenter";
|
private static final String CITY_ROOM_PREFIX = "city_";
|
|
/** 是否启用 mock 车辆（支持 Nacos 动态刷新） */
|
@Value("${command-center.mock.enabled:true}")
|
private boolean mockEnabled;
|
|
@Resource
|
private SocketIOServer socketIOServer;
|
@Resource
|
private CommandCenterService commandCenterService;
|
@Resource
|
private PassengerCallService passengerCallService;
|
@Resource
|
private SocketIOSessionManager sessionManager;
|
|
/** 车辆推送周期（每次重新聚合一次）的最近 KPI 缓存，供 5s KPI 推送复用 */
|
private final Map<String, CommandCenterKpiVo> latestKpiByCity = new ConcurrentHashMap<>();
|
|
/**
|
* 每 2 秒聚合一次车辆并向已订阅 Room 推送车辆位置；同时缓存 KPI 给 5s 推送复用
|
*/
|
@Scheduled(fixedDelay = 2000)
|
public void pushVehicles() {
|
Set<String> activeCityCodes = commandCenterService.getActiveRoomCityCodes();
|
if (activeCityCodes.isEmpty()) {
|
return;
|
}
|
Map<String, AggregateResult> resultMap;
|
try {
|
resultMap = commandCenterService.aggregateAllCities(activeCityCodes);
|
} catch (Exception e) {
|
log.warn("[CommandCenter] aggregateAllCities unexpected failure, use mock fallback if enabled", e);
|
resultMap = Collections.emptyMap();
|
}
|
for (String cityCode : activeCityCodes) {
|
try {
|
AggregateResult result = resultMap.get(cityCode);
|
List<VehicleMonitorVo> vehicles = result != null && result.getVehicles() != null
|
? new ArrayList<>(result.getVehicles()) : new ArrayList<>();
|
CommandCenterKpiVo kpi = result != null ? result.getKpi() : null;
|
|
if (mockEnabled) {
|
vehicles = appendMockVehicles(vehicles);
|
if (CommandCenterMockKpiSupport.isEmptyKpi(kpi)) {
|
kpi = CommandCenterMockKpiSupport.buildMockKpi();
|
}
|
}
|
latestKpiByCity.put(cityCode, kpi != null ? kpi : new CommandCenterKpiVo());
|
|
BroadcastOperations room = socketIOServer.getNamespace(NAMESPACE)
|
.getRoomOperations(CITY_ROOM_PREFIX + cityCode);
|
if (room.getClients().isEmpty()) {
|
continue;
|
}
|
room.sendEvent("vehicles", vehicles);
|
} catch (Exception e) {
|
log.error("pushVehicles failed for cityCode={}", cityCode, e);
|
if (mockEnabled) {
|
pushMockVehiclesFallback(cityCode);
|
}
|
}
|
}
|
latestKpiByCity.keySet().retainAll(activeCityCodes);
|
}
|
|
/**
|
* mock 模式下聚合异常时的兜底推送
|
*/
|
private void pushMockVehiclesFallback(String cityCode) {
|
try {
|
BroadcastOperations room = socketIOServer.getNamespace(NAMESPACE)
|
.getRoomOperations(CITY_ROOM_PREFIX + cityCode);
|
if (room.getClients().isEmpty()) {
|
return;
|
}
|
List<VehicleMonitorVo> vehicles = appendMockVehicles(Collections.emptyList());
|
latestKpiByCity.put(cityCode, CommandCenterMockKpiSupport.buildMockKpi());
|
room.sendEvent("vehicles", vehicles);
|
} catch (Exception ex) {
|
log.error("pushMockVehiclesFallback failed for cityCode={}", cityCode, ex);
|
}
|
}
|
|
/**
|
* 每 5 秒推送 KPI、呼单与天气
|
*/
|
@Scheduled(fixedDelay = 5000)
|
public void pushKpiAndCalls() {
|
Set<String> activeCityCodes = commandCenterService.getActiveRoomCityCodes();
|
if (activeCityCodes.isEmpty()) {
|
return;
|
}
|
for (String cityCode : activeCityCodes) {
|
try {
|
BroadcastOperations room = socketIOServer.getNamespace(NAMESPACE)
|
.getRoomOperations(CITY_ROOM_PREFIX + cityCode);
|
if (room.getClients().isEmpty()) {
|
continue;
|
}
|
CommandCenterKpiVo kpi = latestKpiByCity.get(cityCode);
|
if (mockEnabled && CommandCenterMockKpiSupport.isEmptyKpi(kpi)) {
|
kpi = CommandCenterMockKpiSupport.buildMockKpi();
|
}
|
List<PassengerCallVo> calls;
|
try {
|
calls = passengerCallService.listByCityCode(cityCode);
|
} catch (Exception e) {
|
log.warn("[CommandCenter] listByCityCode failed city={}", cityCode, e);
|
calls = Collections.emptyList();
|
}
|
WeatherDTO.Lives weather = commandCenterService.fetchWeather(cityCode);
|
if (kpi != null) {
|
room.sendEvent("kpi", kpi);
|
}
|
if (mockEnabled) {
|
calls = appendMockPassengerCalls(calls);
|
}
|
room.sendEvent("passenger_calls", calls);
|
if (weather != null) {
|
room.sendEvent("weather", weather);
|
}
|
} catch (Exception e) {
|
log.error("pushKpiAndCalls failed for cityCode={}", cityCode, e);
|
if (mockEnabled) {
|
pushMockKpiAndCallsFallback(cityCode);
|
}
|
}
|
}
|
}
|
|
private void pushMockKpiAndCallsFallback(String cityCode) {
|
try {
|
BroadcastOperations room = socketIOServer.getNamespace(NAMESPACE)
|
.getRoomOperations(CITY_ROOM_PREFIX + cityCode);
|
if (room.getClients().isEmpty()) {
|
return;
|
}
|
room.sendEvent("kpi", CommandCenterMockKpiSupport.buildMockKpi());
|
room.sendEvent("passenger_calls", appendMockPassengerCalls(Collections.emptyList()));
|
} catch (Exception ex) {
|
log.error("pushMockKpiAndCallsFallback failed for cityCode={}", cityCode, ex);
|
}
|
}
|
|
/**
|
* 每 2 秒向已订阅详情的客户端推送单车详情（每辆车只查一次）
|
*/
|
@Scheduled(fixedDelay = 2000)
|
public void pushVehicleDetails() {
|
Set<Long> vehicleIds = sessionManager.getSubscribedVehicleIds();
|
if (vehicleIds.isEmpty()) {
|
return;
|
}
|
for (Long vehicleId : vehicleIds) {
|
try {
|
CommandCenterVehicleDetailVo detail = resolveVehicleDetail(vehicleId);
|
if (detail == null) {
|
continue;
|
}
|
pushVehicleDetailToSubscribers(vehicleId, detail);
|
} catch (Exception e) {
|
log.warn("[CommandCenter] pushVehicleDetails failed vehicleId={}", vehicleId, e);
|
if (mockEnabled && CommandCenterMockDetailSupport.isMockVehicleId(vehicleId)) {
|
CommandCenterVehicleDetailVo mockDetail =
|
CommandCenterMockDetailSupport.buildMockVehicleDetail(vehicleId);
|
pushVehicleDetailToSubscribers(vehicleId, mockDetail);
|
}
|
}
|
}
|
}
|
|
private CommandCenterVehicleDetailVo resolveVehicleDetail(Long vehicleId) {
|
if (mockEnabled && CommandCenterMockDetailSupport.isMockVehicleId(vehicleId)) {
|
return CommandCenterMockDetailSupport.buildMockVehicleDetail(vehicleId);
|
}
|
// 优先从 2s 聚合缓存获取（零 RPC），未命中时降级为完整 RPC 查询
|
CommandCenterVehicleDetailVo detail = commandCenterService.getVehicleDetailFromCache(vehicleId);
|
if (detail != null) {
|
return detail;
|
}
|
return commandCenterService.getVehicleDetailForPush(vehicleId);
|
}
|
|
private void pushVehicleDetailToSubscribers(Long vehicleId, CommandCenterVehicleDetailVo detail) {
|
if (detail == null) {
|
return;
|
}
|
for (String sessionId : sessionManager.getSubscriberSessionIds(vehicleId)) {
|
SocketIOClient client = sessionManager.getClient(sessionId);
|
if (client != null && client.isChannelOpen()) {
|
client.sendEvent("vehicle_detail", detail);
|
}
|
}
|
}
|
|
// ==================== Mock 辅助方法 ====================
|
|
private List<PassengerCallVo> appendMockPassengerCalls(List<PassengerCallVo> realCalls) {
|
List<PassengerCallVo> merged = new ArrayList<>(realCalls != null ? realCalls : new ArrayList<>());
|
|
PassengerCallVo mock1 = new PassengerCallVo();
|
mock1.setOrderId(88881L);
|
mock1.setAppUserId(231313L);
|
mock1.setOrderNumber("MOCK_ORDER_001");
|
mock1.setStartLon(113.3245);
|
mock1.setStartLat(23.1291);
|
mock1.setStartAddress("广州塔");
|
mock1.setEndLon(113.3312);
|
mock1.setEndLat(23.1370);
|
mock1.setEndAddress("珠江新城");
|
mock1.setCreateTime(LocalDateTime.now().minusSeconds(45));
|
mock1.setWaitingSeconds(45L);
|
|
PassengerCallVo mock2 = new PassengerCallVo();
|
mock2.setOrderId(88882L);
|
mock2.setAppUserId(33333333L);
|
mock2.setOrderNumber("MOCK_ORDER_002");
|
mock2.setStartLon(113.3180);
|
mock2.setStartLat(23.1205);
|
mock2.setStartAddress("客村地铁站");
|
mock2.setEndLon(113.3400);
|
mock2.setEndLat(23.1320);
|
mock2.setEndAddress("琶洲会展中心");
|
mock2.setCreateTime(LocalDateTime.now().minusSeconds(120));
|
mock2.setWaitingSeconds(120L);
|
|
merged.addAll(Arrays.asList(mock1, mock2));
|
return merged;
|
}
|
|
private List<VehicleMonitorVo> appendMockVehicles(List<VehicleMonitorVo> realVehicles) {
|
List<VehicleMonitorVo> merged = new ArrayList<>(realVehicles != null ? realVehicles : Collections.emptyList());
|
|
merged.add(buildMockVehicle(99901L, "MOCK_VIN_CRUISING", "粤A MOCK01",
|
"CRUISING", "巡游中", "CRUISING",
|
113.3245, 23.1091, 30.0f, 85, false));
|
merged.add(buildMockVehicle(99902L, "MOCK_VIN_PICKUP", "粤A MOCK02",
|
"HAS_ORDER", "接驾中", "PICK_UP",
|
113.3210, 23.1190, 35.0f, 72, false));
|
merged.add(buildMockVehicle(99903L, "MOCK_VIN_DROPOFF", "粤A MOCK03",
|
"HAS_ORDER", "送驾中", "DROP_OFF",
|
113.3500, 23.1020, 40.0f, 65, false));
|
merged.add(buildMockVehicle(99904L, "MOCK_VIN_CHARGING", "粤A MOCK04",
|
"OCCUPIED", "充电中", "CHARGING",
|
113.3180, 23.0985, 0.0f, 45, true));
|
merged.add(buildMockVehicle(99905L, "MOCK_VIN_PREPARING", "粤A MOCK05",
|
"OCCUPIED", "整备中", "PREPARING",
|
113.3150, 23.1380, 0.0f, 90, false));
|
merged.add(buildMockVehicle(99906L, "MOCK_VIN_RETURNING", "粤A MOCK06",
|
"OCCUPIED", "回场中", "RETURNING",
|
113.2690, 23.0028, 28.0f, 55, false));
|
merged.add(buildMockVehicle(99907L, "MOCK_VIN_ABNORMAL", "粤A MOCK07",
|
"ABNORMAL", "车辆异常", "ABNORMAL",
|
113.2730, 23.1650, 0.0f, 30, false));
|
merged.add(buildMockVehicle(99908L, "MOCK_VIN_ODDOUT", "粤A MOCK08",
|
"ABNORMAL", "ODD越界", "ODD_OUT",
|
113.3560, 23.1270, 15.0f, 60, false));
|
|
return merged;
|
}
|
|
private VehicleMonitorVo buildMockVehicle(Long vehicleId, String vin, String plateNumber,
|
String monitorGroup, String monitorStatus, String stateCode,
|
double lon, double lat, float speed, int electricity, boolean charging) {
|
VehicleMonitorVo mock = new VehicleMonitorVo();
|
mock.setVehicleId(vehicleId);
|
mock.setVin(vin);
|
mock.setPlateNumber(plateNumber);
|
mock.setMonitorGroup(monitorGroup);
|
mock.setMonitorStatus(monitorStatus);
|
mock.setStateCode(stateCode);
|
mock.setLon(lon);
|
mock.setLat(lat);
|
mock.setCourse(0.0);
|
mock.setSpeed(speed);
|
mock.setCurrentElectricity(electricity);
|
mock.setCharging(charging);
|
mock.setPositionUpdateTime(System.currentTimeMillis());
|
return mock;
|
}
|
}
|
