1

njzscloud-common/njzscloud-common-localizer/src/main/java/com/njzscloud/common/localizer/tuqiang/listener/TuqiangListeners.java

@@ -44,7 +44,7 @@ public class TuqiangListeners {
         RealtimeLocationResult realtimeLocationResult = BeanUtil.copyProperties(locationReportMsg, RealtimeLocationResult.class)
                 .setTerminalId(terminalId);
         Mqtt.publish(terminalId + "/track_location", realtimeLocationResult);
-
+        Mqtt.publish(terminalId + "/track_location_real", realtimeLocationResult);
     }
 
     /**
@@ -65,6 +65,7 @@ public class TuqiangListeners {
                     .setTerminalId(terminalId)
                     .setType(type);
             Mqtt.publish(terminalId + "/track_location", realtimeLocationResult);
+            Mqtt.publish(terminalId + "/track_location_real", realtimeLocationResult);
         }
     }
 

njzscloud-svr/src/main/java/com/njzscloud/supervisory/biz/mapper/TruckLocationTrackMapper.java

@@ -3,6 +3,8 @@ package com.njzscloud.supervisory.biz.mapper;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.njzscloud.supervisory.biz.pojo.entity.TruckLocationTrackEntity;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
 
 /**
  * 车辆定位数据
@@ -10,4 +12,10 @@ import org.apache.ibatis.annotations.Mapper;
 @Mapper
 public interface TruckLocationTrackMapper extends BaseMapper<TruckLocationTrackEntity> {
 
+    @Select("SELECT b.gps\n" +
+            "FROM order_info a\n" +
+            "         INNER JOIN biz_truck b ON b.id = a.truck_id AND b.deleted = 0\n" +
+            "WHERE a.id = #{orderId}\n" +
+            "  AND a.deleted = 0")
+    String gpsId(@Param("orderId") Long orderId);
 }

njzscloud-svr/src/main/java/com/njzscloud/supervisory/biz/service/TruckLocationTrackService.java

@@ -11,16 +11,18 @@ import com.baomidou.mybatisplus.extension.service.IService;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njzscloud.common.mp.support.PageParam;
 import com.njzscloud.common.mp.support.PageResult;
+import com.njzscloud.common.mqtt.support.MqttMsg;
 import com.njzscloud.supervisory.biz.mapper.TruckLocationTrackMapper;
 import com.njzscloud.supervisory.biz.pojo.entity.TruckLocationTrackEntity;
+import com.njzscloud.supervisory.order.pojo.result.RealtimeLocationResult;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import java.io.IOException;
-import java.time.Duration;
 import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -36,9 +38,8 @@ import java.util.concurrent.locks.ReentrantLock;
 @Slf4j
 @Service
 public class TruckLocationTrackService extends ServiceImpl<TruckLocationTrackMapper, TruckLocationTrackEntity> implements IService<TruckLocationTrackEntity> {
-
     ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
-    HashMap<Long, List<TruckLocationTrackEntity>> LOCATION_CACHE = new HashMap<>();
+    HashMap<String, List<SseEmitter>> emitters = new HashMap<>();
     ReentrantLock lock = new ReentrantLock();
 
 
@@ -274,6 +275,7 @@ public class TruckLocationTrackService extends ServiceImpl<TruckLocationTrackMap
         return emitter;
     }
 
+    /*
     public void cache(TruckLocationTrackEntity truckLocationTrackEntity) {
         Long orderId = truckLocationTrackEntity.getOrderId();
         lock.lock();
@@ -287,11 +289,17 @@ public class TruckLocationTrackService extends ServiceImpl<TruckLocationTrackMap
             lock.unlock();
         }
     }
+    */
 
     public SseEmitter realtime(Long orderId) {
+        String gpsId = baseMapper.gpsId(orderId);
         SseEmitter emitter = new SseEmitter(0L);
+        lock.lock();
+        List<SseEmitter> value = emitters.computeIfAbsent(gpsId, k -> new LinkedList<>());
+        value.add(emitter);
+        lock.unlock();
 
-        // 提交异步任务
+       /*  // 提交异步任务
         CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
             try {
                 // 记录上一条数据的时间，用于计算间隔
@@ -344,26 +352,61 @@ public class TruckLocationTrackService extends ServiceImpl<TruckLocationTrackMap
                 // 其他异常
                 emitter.completeWithError(e);
             }
-        }, threadPoolExecutor);
+        }, threadPoolExecutor); */
 
         // 注册连接关闭回调，中断任务
         // 注册连接关闭回调，中断任务
         emitter.onCompletion(() -> {
             log.info("连接关闭，取消任务");
-            future.cancel(true);
+            lock.lock();
+            List<SseEmitter> emitterList = emitters.get(gpsId);
+            emitterList.remove(emitter);
+            lock.unlock();
         });
         emitter.onTimeout(() -> {
             log.info("连接超时，取消任务");
-            future.cancel(true);
+            lock.lock();
+            List<SseEmitter> emitterList = emitters.get(gpsId);
+            emitterList.remove(emitter);
+            lock.unlock();
         });
         emitter.onError((e) -> {
             log.info("连接错误，取消任务");
-            future.cancel(true);
+            lock.lock();
+            List<SseEmitter> emitterList = emitters.get(gpsId);
+            emitterList.remove(emitter);
+            lock.unlock();
         });
 
-
         return emitter;
     }
 
+    public void sendRealTimeData(MqttMsg msg) {
+        RealtimeLocationResult realtimeLocationResult = msg.getMsg(RealtimeLocationResult.class);
+        if (realtimeLocationResult != null) {
+            String gpsId = realtimeLocationResult.getTerminalId();
+            lock.lock();
+            List<SseEmitter> emitterList = emitters.get(gpsId);
+            lock.unlock();
+            if (CollUtil.isEmpty(emitterList)) {
+                log.info("未找到订单 {} 的实时数据订阅者", gpsId);
+                return;
+            }
+
+            for (SseEmitter emitter : emitterList) {
+                send(emitter, "realtimeData", new TruckLocationTrackEntity()
+                        .setId(IdUtil.getSnowflakeNextId())
+                        .setTerminalId(gpsId)
+                        .setLatitude(realtimeLocationResult.getLatitude())
+                        .setLongitude(realtimeLocationResult.getLongitude())
+                        .setAltitude(realtimeLocationResult.getAltitude())
+                        .setSpeed(realtimeLocationResult.getSpeed())
+                        .setLocationTime(LocalDateTime.parse(realtimeLocationResult.getTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))
+                        .setDirection(realtimeLocationResult.getDirection())
+                        .setOverspeed(realtimeLocationResult.isOverspeed())
+                        .setCompensate(realtimeLocationResult.getType() == 1));
+            }
+        }
+    }
 
 }

njzscloud-svr/src/main/java/com/njzscloud/supervisory/device/service/DeviceLocalizerService.java

@@ -7,10 +7,12 @@ import com.njzscloud.common.mp.support.PageParam;
 import com.njzscloud.common.mp.support.PageResult;
 import com.njzscloud.common.mqtt.support.MqttMsg;
 import com.njzscloud.common.mqtt.util.Mqtt;
+import com.njzscloud.supervisory.biz.service.TruckLocationTrackService;
 import com.njzscloud.supervisory.device.mapper.DeviceLocalizerMapper;
 import com.njzscloud.supervisory.device.pojo.entity.DeviceLocalizerEntity;
 import com.njzscloud.supervisory.device.pojo.result.DeviceInfoResult;
 import com.njzscloud.supervisory.device.pojo.result.HeartbeatResult;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.event.EventListener;
@@ -25,7 +27,9 @@ import java.util.List;
  */
 @Slf4j
 @Service
+@RequiredArgsConstructor
 public class DeviceLocalizerService extends ServiceImpl<DeviceLocalizerMapper, DeviceLocalizerEntity> implements IService<DeviceLocalizerEntity> {
+    private final TruckLocationTrackService truckLocationTrackService;
 
     /**
      * 新增
@@ -70,6 +74,7 @@ public class DeviceLocalizerService extends ServiceImpl<DeviceLocalizerMapper, D
             String terminalId = localizerEntity.getTerminalId();
             Mqtt.subscribe(terminalId + "/online", this::heartbeat);
             Mqtt.subscribe(terminalId + "/device_info", this::updateDeviceLocalizerStatus);
+            Mqtt.subscribe(terminalId + "/track_location_real", truckLocationTrackService::sendRealTimeData);
         }
     }
 

njzscloud-svr/src/main/java/com/njzscloud/supervisory/order/service/OrderInfoService.java

@@ -672,11 +672,11 @@ public class OrderInfoService extends ServiceImpl<OrderInfoMapper, OrderInfoEnti
                         .setOverspeed(realtimeLocationResult.isOverspeed())
                         .setCompensate(realtimeLocationResult.getType() == 1);
                 truckLocationTrackService.save(entity);
-                truckLocationTrackService.cache(entity);
+                // truckLocationTrackService.cache(entity);
             });
             Mqtt.publish("location/track", MapUtil.builder()
                     .put("terminalId", gpsId)
-                    .put("interval", 3)
+                    .put("interval", 1)
                     .build());
         }).whenComplete((aVoid, throwable) -> {
             if (throwable != null) {

njzscloud-svr/src/main/resources/application-dev.yml

@@ -56,15 +56,19 @@ mybatis-plus:
       port: 33061
 
 wechat:
-  app-id: wx989ea47a5ddf9bfb
+  app-id: wx3c06d9dd4e56c58d
-  app-secret: 66c98dc487a372acb4f1931b38fee8ff
+  app-secret: ff280a71a4c06fc2956178f8c472ef96
+  #  app-id: wx989ea47a5ddf9bfb
+  #  app-secret: 66c98dc487a372acb4f1931b38fee8ff
   base-url: https://api.weixin.qq.com
   pay:
     # 子商户配置
-    sub-app-id: wx989ea47a5ddf9bfb
+    sub-app-id: wx3c06d9dd4e56c58d
+    #    sub-app-id: wx989ea47a5ddf9bfb
     sub-mch-id: 1900000100
     # API密钥（32位字符串）
-    api-key: 66c98dc487a372acb4f1931b38fee8ff
+    api-key: your-32-character-api-key-here
+    #    api-key: 66c98dc487a372acb4f1931b38fee8ff
     # 证书序列号
     cert-serial-no: your-cert-serial-number
     # 私钥文件路径

njzscloud-svr/src/main/resources/mapper/order/OrderInfoMapper.xml

@@ -121,7 +121,9 @@
                m.contacts station_contacts,
                m.phone    station_phone,
                a.contacts,
-               a.phone
+               a.phone,
+               a.expect_time,
+               a.estimated_quantity
         FROM order_info a
                  LEFT JOIN order_cargo_place b ON b.id = a.cargo_place_id
                  LEFT JOIN order_goods c ON c.id = a.goods_id