feat(location): 添加骑手位置实时推送功能

- 新增 AdminLocationController 提供管理端位置接口 - 创建 AdminRiderLocationVO 和 LocationPushMessage 数据对象 - 实现 LocationPushService 和 LocationSessionRegistry 推送服务 - 集成 WebSocket 配置支持实时位置更新 - 在 RiderLocationService 中添加按城市查询活跃骑手功能 - 实现位置更新时自动推送给对应城市的 WebSocket 客户端 - 添加 WebSocket 握手拦截器验证管理员身份和权限

feat(location): 添加骑手位置实时推送功能
- 新增 AdminLocationController 提供管理端位置接口 - 创建 AdminRiderLocationVO 和 LocationPushMessage 数据对象 - 实现 LocationPushService 和 LocationSessionRegistry 推送服务 - 集成 WebSocket 配置支持实时位置更新 - 在 RiderLocationService 中添加按城市查询活跃骑手功能 - 实现位置更新时自动推送给对应城市的 WebSocket 客户端 - 添加 WebSocket 握手拦截器验证管理员身份和权限
蒋勇
1 parent dff9b9d0
Showing 11 changed files with 388 additions and 0 deletions
pom.xml
src/main/java/com/diligrp/rider/config/LocationWebSocketConfig.java
src/main/java/com/diligrp/rider/controller/AdminLocationController.java
src/main/java/com/diligrp/rider/service/RiderLocationService.java
src/main/java/com/diligrp/rider/service/impl/RiderLocationServiceImpl.java
src/main/java/com/diligrp/rider/vo/AdminRiderLocationVO.java
src/main/java/com/diligrp/rider/websocket/LocationPushMessage.java
src/main/java/com/diligrp/rider/websocket/LocationPushService.java
src/main/java/com/diligrp/rider/websocket/LocationSessionRegistry.java
src/main/java/com/diligrp/rider/websocket/LocationWebSocketHandler.java
src/main/java/com/diligrp/rider/websocket/LocationWebSocketHandshakeInterceptor.java
@@ -28,6 +28,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
         <!-- MyBatis-Plus (Spring Boot 3 专用 starter) -->
         <dependency>
+package com.diligrp.rider.config;
+
+import com.diligrp.rider.websocket.LocationWebSocketHandshakeInterceptor;
+import com.diligrp.rider.websocket.LocationWebSocketHandler;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
+
+@Configuration
+@EnableWebSocket
+@RequiredArgsConstructor
+public class LocationWebSocketConfig implements WebSocketConfigurer {
+
+    private final LocationWebSocketHandler locationWebSocketHandler;
+    private final LocationWebSocketHandshakeInterceptor locationWebSocketHandshakeInterceptor;
+
+    @Override
+    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
+        registry.addHandler(locationWebSocketHandler, "/ws/location")
+                .addInterceptors(locationWebSocketHandshakeInterceptor)
+                .setAllowedOriginPatterns("*");
+    }
+}
+package com.diligrp.rider.controller;
+
+import com.diligrp.rider.common.exception.BizException;
+import com.diligrp.rider.common.result.Result;
+import com.diligrp.rider.service.RiderLocationService;
+import com.diligrp.rider.vo.AdminRiderLocationVO;
+import jakarta.servlet.http.HttpServletRequest;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("/api/admin/location")
+@RequiredArgsConstructor
+public class AdminLocationController {
+
+    private final RiderLocationService riderLocationService;
+
+    @GetMapping("/active")
+    public Result<List<AdminRiderLocationVO>> active(@RequestParam(required = false) Long cityId,
+                                                     HttpServletRequest request) {
+        String role = (String) request.getAttribute("role");
+        Long resolvedCityId = "substation".equals(role)
+                ? (Long) request.getAttribute("cityId")
+                : cityId;
+        if (resolvedCityId == null || resolvedCityId < 1) {
+            throw new BizException(400, "请选择城市");
+        }
+        return Result.success(riderLocationService.listActiveByCity(resolvedCityId));
+    }
+}
 package com.diligrp.rider.service;
 import com.diligrp.rider.dto.LocationDTO;
+import com.diligrp.rider.vo.AdminRiderLocationVO;
 import com.diligrp.rider.vo.NearbyRiderVO;
 import java.math.BigDecimal;
@@ -13,6 +14,8 @@ public interface RiderLocationService {
     /** 获取骑手位置 */
     LocationDTO getLocation(Long riderId);
+    List<AdminRiderLocationVO> listActiveByCity(Long cityId);
+
     /**
      * 获取附近在线骑手列表
      *
@@ -10,7 +10,9 @@ import com.diligrp.rider.mapper.RiderMapper;
 import com.diligrp.rider.service.CityService;
 import com.diligrp.rider.service.RiderLocationService;
 import com.diligrp.rider.util.GeoUtil;
+import com.diligrp.rider.vo.AdminRiderLocationVO;
 import com.diligrp.rider.vo.NearbyRiderVO;
+import com.diligrp.rider.websocket.LocationPushService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -29,6 +31,7 @@ public class RiderLocationServiceImpl implements RiderLocationService {
     private final RiderLocationMapper locationMapper;
     private final RiderMapper riderMapper;
     private final CityService cityService;
+    private final LocationPushService locationPushService;
     /**
      * 按骑手维度更新最新位置。
@@ -38,11 +41,22 @@ public class RiderLocationServiceImpl implements RiderLocationService {
         long now = System.currentTimeMillis() / 1000;
         log.debug("更新骑手位置，参数 riderId={} lng={} lat={} updateTime={}",
                 riderId, dto.getLng(), dto.getLat(), now);
+        Rider rider = riderMapper.selectById(riderId);
         locationMapper.upsertLocation(
                 riderId,
                 dto.getLng().stripTrailingZeros().toPlainString(),
                 dto.getLat().stripTrailingZeros().toPlainString(),
                 now);
+        if (rider != null && rider.getCityId() != null) {
+            AdminRiderLocationVO location = new AdminRiderLocationVO();
+            location.setRiderId(riderId);
+            location.setRiderName(rider.getUserNickname());
+            location.setCityId(rider.getCityId());
+            location.setLng(dto.getLng());
+            location.setLat(dto.getLat());
+            location.setUpdateTime(now);
+            locationPushService.pushRiderLocation(location);
+        }
     }
     /**
@@ -63,6 +77,54 @@ public class RiderLocationServiceImpl implements RiderLocationService {
         return dto;
     }
+    @Override
+    public List<AdminRiderLocationVO> listActiveByCity(Long cityId) {
+        List<AdminRiderLocationVO> result = new ArrayList<>();
+        if (cityId == null || cityId < 1) {
+            return result;
+        }
+
+        List<Rider> riders = riderMapper.selectList(new LambdaQueryWrapper<Rider>()
+                .eq(Rider::getCityId, cityId)
+                .eq(Rider::getIsRest, 0)
+                .eq(Rider::getUserStatus, 1));
+        if (riders.isEmpty()) {
+            return result;
+        }
+
+        List<Long> riderIds = riders.stream().map(Rider::getId).toList();
+        long minUpdateTime = System.currentTimeMillis() / 1000 - ACTIVE_WINDOW_SECONDS;
+        List<RiderLocation> locations = locationMapper.selectList(
+                new LambdaQueryWrapper<RiderLocation>()
+                        .in(RiderLocation::getUid, riderIds)
+                        .ge(RiderLocation::getUpdateTime, minUpdateTime));
+        if (locations.isEmpty()) {
+            return result;
+        }
+
+        java.util.Map<Long, Rider> riderMap = riders.stream()
+                .collect(java.util.stream.Collectors.toMap(Rider::getId, rider -> rider));
+        for (RiderLocation location : locations) {
+            Rider rider = riderMap.get(location.getUid());
+            if (rider == null) {
+                continue;
+            }
+            try {
+                AdminRiderLocationVO vo = new AdminRiderLocationVO();
+                vo.setRiderId(rider.getId());
+                vo.setRiderName(rider.getUserNickname());
+                vo.setCityId(rider.getCityId());
+                vo.setLng(new BigDecimal(location.getLng()));
+                vo.setLat(new BigDecimal(location.getLat()));
+                vo.setUpdateTime(location.getUpdateTime());
+                result.add(vo);
+            } catch (Exception ignored) {
+            }
+        }
+        log.debug("查询城市活跃骑手位置完成，cityId={} count={}", cityId, result.size());
+        return result;
+    }
+
     /**
      * 查询城市范围内最近活跃的附近骑手。
      */
+package com.diligrp.rider.vo;
+
+import lombok.Data;
+
+import java.math.BigDecimal;
+
+@Data
+public class AdminRiderLocationVO {
+    private Long riderId;
+    private String riderName;
+    private BigDecimal lng;
+    private BigDecimal lat;
+    private Long cityId;
+    private Long updateTime;
+}
+package com.diligrp.rider.websocket;
+
+import lombok.Data;
+
+import java.math.BigDecimal;
+
+@Data
+public class LocationPushMessage {
+    private String type;
+    private Long riderId;
+    private String riderName;
+    private Long cityId;
+    private BigDecimal lng;
+    private BigDecimal lat;
+    private Long updateTime;
+}
+package com.diligrp.rider.websocket;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.diligrp.rider.vo.AdminRiderLocationVO;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class LocationPushService {
+
+    private final LocationSessionRegistry sessionRegistry;
+    private final ObjectMapper objectMapper;
+
+    public void pushRiderLocation(AdminRiderLocationVO location) {
+        if (location == null || location.getCityId() == null) {
+            return;
+        }
+        LocationPushMessage message = new LocationPushMessage();
+        message.setType("rider_location_update");
+        message.setRiderId(location.getRiderId());
+        message.setRiderName(location.getRiderName());
+        message.setCityId(location.getCityId());
+        message.setLng(location.getLng());
+        message.setLat(location.getLat());
+        message.setUpdateTime(location.getUpdateTime());
+        broadcast(location.getCityId(), message);
+    }
+
+    public void pushConnected(Long cityId) {
+        if (cityId == null) {
+            return;
+        }
+        LocationPushMessage message = new LocationPushMessage();
+        message.setType("connected");
+        message.setCityId(cityId);
+        broadcast(cityId, message);
+    }
+
+    private void broadcast(Long cityId, LocationPushMessage message) {
+        try {
+            String payload = objectMapper.writeValueAsString(message);
+            for (WebSocketSession session : sessionRegistry.getSessions(cityId)) {
+                if (!session.isOpen()) {
+                    sessionRegistry.unregister(session);
+                    continue;
+                }
+                synchronized (session) {
+                    session.sendMessage(new TextMessage(payload));
+                }
+            }
+            log.debug("骑手位置推送完成，cityId={} type={}", cityId, message.getType());
+        } catch (Exception e) {
+            log.warn("骑手位置推送失败，cityId={} type={}", cityId, message.getType(), e);
+        }
+    }
+}
+package com.diligrp.rider.websocket;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.WebSocketSession;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class LocationSessionRegistry {
+
+    private final ConcurrentHashMap<Long, Set<WebSocketSession>> citySessions = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, Long> sessionCities = new ConcurrentHashMap<>();
+
+    public void register(Long cityId, WebSocketSession session) {
+        citySessions.computeIfAbsent(cityId, key -> ConcurrentHashMap.newKeySet()).add(session);
+        sessionCities.put(session.getId(), cityId);
+    }
+
+    public void unregister(WebSocketSession session) {
+        Long cityId = sessionCities.remove(session.getId());
+        if (cityId == null) {
+            return;
+        }
+        Set<WebSocketSession> sessions = citySessions.get(cityId);
+        if (sessions == null) {
+            return;
+        }
+        sessions.remove(session);
+        if (sessions.isEmpty()) {
+            citySessions.remove(cityId, sessions);
+        }
+    }
+
+    public Set<WebSocketSession> getSessions(Long cityId) {
+        return citySessions.getOrDefault(cityId, Collections.emptySet());
+    }
+}
+package com.diligrp.rider.websocket;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class LocationWebSocketHandler extends TextWebSocketHandler {
+
+    private final LocationSessionRegistry sessionRegistry;
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
+        Object cityIdObj = session.getAttributes().get("cityId");
+        if (!(cityIdObj instanceof Long cityId)) {
+            session.close(CloseStatus.NOT_ACCEPTABLE.withReason("缺少城市信息"));
+            return;
+        }
+        sessionRegistry.register(cityId, session);
+        log.debug("位置推送连接建立，sessionId={} cityId={}", session.getId(), cityId);
+    }
+
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
+        if ("ping".equalsIgnoreCase(message.getPayload())) {
+            session.sendMessage(new TextMessage("pong"));
+        }
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
+        sessionRegistry.unregister(session);
+        log.debug("位置推送连接关闭，sessionId={} code={}", session.getId(), status.getCode());
+    }
+
+    @Override
+    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
+        sessionRegistry.unregister(session);
+        if (session.isOpen()) {
+            session.close(CloseStatus.SERVER_ERROR);
+        }
+        log.warn("位置推送连接异常，sessionId={}", session.getId(), exception);
+    }
+}
+package com.diligrp.rider.websocket;
+
+import com.diligrp.rider.common.exception.BizException;
+import com.diligrp.rider.config.JwtUtil;
+import com.diligrp.rider.entity.Substation;
+import com.diligrp.rider.mapper.SubstationMapper;
+import io.jsonwebtoken.Claims;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.server.ServerHttpRequest;
+import org.springframework.http.server.ServerHttpResponse;
+import org.springframework.http.server.ServletServerHttpRequest;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+import org.springframework.web.socket.WebSocketHandler;
+import org.springframework.web.socket.server.HandshakeInterceptor;
+
+import java.util.Map;
+
+@Component
+@RequiredArgsConstructor
+public class LocationWebSocketHandshakeInterceptor implements HandshakeInterceptor {
+
+    private final JwtUtil jwtUtil;
+    private final SubstationMapper substationMapper;
+
+    @Override
+    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
+                                   WebSocketHandler wsHandler, Map<String, Object> attributes) {
+        if (!(request instanceof ServletServerHttpRequest servletRequest)) {
+            return false;
+        }
+        String token = servletRequest.getServletRequest().getParameter("token");
+        if (!StringUtils.hasText(token)) {
+            token = request.getHeaders().getFirst("Authorization");
+        }
+        if (!StringUtils.hasText(token)) {
+            throw new BizException(700, "请先登录");
+        }
+        if (token.startsWith("Bearer ")) {
+            token = token.substring(7);
+        }
+
+        Claims claims = jwtUtil.getAdminClaims(token);
+        Object adminIdObj = claims.get("adminId");
+        String role = claims.get("role", String.class);
+        if (!(adminIdObj instanceof Number) || !StringUtils.hasText(role)) {
+            throw new BizException(700, "登录状态失效，请重新登录");
+        }
+
+        Long cityId;
+        Long adminId = ((Number) adminIdObj).longValue();
+        if ("substation".equals(role)) {
+            Substation substation = substationMapper.selectById(adminId);
+            if (substation == null || substation.getCityId() == null) {
+                throw new BizException(700, "分站信息不存在");
+            }
+            cityId = substation.getCityId();
+        } else if ("admin".equals(role)) {
+            String cityIdText = servletRequest.getServletRequest().getParameter("cityId");
+            if (!StringUtils.hasText(cityIdText)) {
+                throw new BizException(400, "管理员连接WebSocket时必须传cityId");
+            }
+            cityId = Long.parseLong(cityIdText);
+        } else {
+            throw new BizException(403, "当前角色无权订阅位置推送");
+        }
+
+        attributes.put("adminId", adminId);
+        attributes.put("role", role);
+        attributes.put("cityId", cityId);
+        return true;
+    }
+
+    @Override
+    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
+                               WebSocketHandler wsHandler, Exception exception) {
+    }
+}