配送平台订单骑手位置服务

conf/script/1000/expressApi/BE_Express_RiderLocation_SFTC.groovy

@@ -0,0 +1,111 @@
+
+import com.alibaba.fastjson2.JSON
+import com.dderp.common.api.BusinessExecutor
+import com.dderp.common.datas.ERPModule
+import com.dderp.common.entity.base.InvokeCallParams
+import com.dderp.common.entity.base.InvokeCallResult
+import com.dderp.common.http.HttpTools
+import com.dySweetFishPlugin.sql.dao.OperatorWait
+import com.sweetfish.service.RetResult
+import groovy.json.JsonSlurper
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+
+import javax.annotation.Resource
+import java.nio.charset.StandardCharsets
+import java.time.LocalDateTime
+import java.time.ZoneOffset
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
+
+class BE_Express_RiderLocation_SFTC implements BusinessExecutor<InvokeCallParams, InvokeCallResult> {
+
+    private final Logger logger = LogManager.getLogger(this.getClass().getSimpleName())
+
+    @Resource(name = "property.sftc.appId")
+    long sfAppId
+
+    @Resource(name = "property.sftc.appKey")
+    String sfAppKey
+
+    @Resource(name = "property.sftc.apiUrl")
+    String sfApiUrl
+
+    @Override
+    String scriptName() {
+        return "顺丰同城订单获取骑手实时坐标"
+    }
+
+    @Override
+    ERPModule module() {
+        return ERPModule.EXPRESS_API
+    }
+
+    @Override
+    OperatorWait getAWait(InvokeCallParams source) {
+        return OperatorWait.AWAIT
+    }
+
+
+    @Override
+    RetResult<InvokeCallParams> checkExecute(InvokeCallParams source) {
+        //检查订单信息
+        def jsonSlurper = new JsonSlurper()
+        def invokeOrder = jsonSlurper.parseText(source.params)
+
+        String orderId = invokeOrder["orderId"] as String
+        //todo 获取订单信息
+
+        return RetResult.<InvokeCallParams> successT().result(source)
+    }
+
+    RetResult<InvokeCallResult> execute(InvokeCallParams source) {
+        //todo
+        def jsonSlurper = new JsonSlurper()
+        def invokeOrder = jsonSlurper.parseText(source.params)
+
+        //转成顺丰需要提交的信息
+        long currentTime = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8))
+
+        def sfOrder = [
+                dev_id   : sfAppId,
+                order_id : invokeOrder["sfOrderId"],
+                shop_id  : invokeOrder[""],
+                push_time: currentTime
+        ]
+
+
+        String postData = JSON.toJSONString(sfOrder)
+
+        logger.info("请求数据: " + postData)
+        String sign = ExpressApiSign.sfGenerateOpenSign(postData, sfAppId, sfAppKey)
+
+        String url = sfApiUrl + "listorderfeed?sign=" + sign
+
+        CompletableFuture<String> apiResult = HttpTools.postHttpContentAsync(url,
+                5000,
+                StandardCharsets.UTF_8,
+                postData,
+                ["Content-Type": "application/json;charset=utf-8"])
+        try {
+            String orderResult = apiResult.get(6000, TimeUnit.SECONDS)
+            //直接输出orderResult一方面unicode编码，一方面有些数据不要，只要结果里面的feed即可这里在转换一下
+
+            def sfOrderJson = jsonSlurper.parseText(orderResult)
+            if (sfOrderJson["error_code"] != 0) {
+                return RetResult.<InvokeCallResult> errorT().retinfo(sfOrderJson["error_msg"] as String)
+            }
+            return RetResult.<InvokeCallResult> successT().result(
+                    InvokeCallResult.success().data(JSON.toJSONString(sfOrderJson["result"]["feed"]))
+            )
+        } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            logger.error(e.getMessage(), e)
+
+            return RetResult.<InvokeCallResult> errorT().retinfo(e.getMessage())
+        }
+    }
+}
+
+

conf/script/1000/orderApi/BE_Order_Sync_RiderLocation_DYLK.groovy

@@ -0,0 +1 @@
+

conf/sqlroot/com/dderp/business/dao/PlatformDao.dql

@@ -24,22 +24,34 @@ order by #{orderBy}
 ;
 
 -- [addPlatformInfo]
-insert into $table$ (id, platformCode, platformName, platformType, requireListSerial,
+insert into $table$ (
+id, platformCode, platformName, platformType,
 voidFlag, createBy, createTime,
 createTimeLong, updateBy, updateTime,
-updateTimeLong ) values ( #{id}, #{platformcode}, #{platformname}, #{platformtype}, #{requirelistserial},
+updateTimeLong )
+values (
+#{id}, #{platformcode}, #{platformname}, #{platformtype},
 #{voidflag}, #{createby}, #{createtime},
 #{createtimelong}, #{updateby}, #{updatetime},
 #{updatetimelong} );
 
 -- [updatePlatformInfo]
-update $table$ set platformName = #{platformname},platformType = #{platformtype},requireListSerial = #{requirelistserial},
+update $table$ set
+platformName = #{platformname},
+platformType = #{platformtype},
+platformCode = #{platformcode},
 updateBy = #{updateby},
-updateTime = #{updatetime},updateTimeLong = #{updatetimelong} where id = #{id};
+updateTime = #{updatetime},
+updateTimeLong = #{updatetimelong}
+where id = #{id};
 
 -- [voidPlatformInfo]
-update $table$ set voidFlag = #{voidflag},updateBy = #{updateby},
-updateTime = #{updatetime},updateTimeLong = #{updatetimelong} where id = #{id};
+update $table$ set
+voidFlag = #{voidflag},
+updateBy = #{updateby},
+updateTime = #{updatetime},
+updateTimeLong = #{updatetimelong}
+where id = #{id};
 
 
 -- [queryPlatformRequireList]

ddBusiness/src/main/java/com/dderp/business/service/flycat/ExpressGeoServiceImpl.java

@@ -0,0 +1,184 @@
+package com.dderp.business.service.flycat;
+
+import com.dderp.common.api.NoSqlKeysService;
+import com.dderp.common.api.PlatformService;
+import com.dderp.common.api.flycat.ExpressGeoService;
+import com.dderp.common.api.flycat.ExpressOutService;
+import com.dderp.common.base.BaseService;
+import com.dderp.common.datas.RedisKeys;
+import com.dderp.common.entity.base.InvokeCallParams;
+import com.dderp.common.entity.base.InvokeCallResult;
+import com.dderp.common.entity.geo.RiderGeoInfo;
+import com.dderp.common.entity.platform.PlatformInfo;
+import com.dderp.common.tool.ERPUtils;
+import com.dySweetFishPlugin.redis.RedisService;
+import com.dySweetFishPlugin.tool.lang.ThreadFactoryWithNamePrefix;
+import com.sweetfish.convert.json.JsonConvert;
+import com.sweetfish.service.Local;
+import com.sweetfish.service.RetResult;
+import com.sweetfish.util.AnyValue;
+import com.sweetfish.util.AutoLoad;
+import com.sweetfish.util.ResourceType;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Resource;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+import java.util.stream.IntStream;
+
+/**
+ * 配送平台坐标服务
+ */
+@AutoLoad(false)
+@Local
+@ResourceType(ExpressGeoService.class)
+public class ExpressGeoServiceImpl extends BaseService implements ExpressGeoService {
+
+    @Resource(name = "property.riderGeoShards")
+    private int riderGeoShards;
+
+    @Resource
+    RedisService redisService;
+
+    @Resource
+    NoSqlKeysService keysService;
+
+    @Resource
+    JsonConvert jsonConvert;
+
+    @Resource
+    PlatformService platformService;
+
+    @Resource
+    ExpressOutService expressOutService;
+
+    private final List<ScheduledThreadPoolExecutor> scheduleScanThreadList = new ArrayList<>();
+
+    private ExecutorService riderWorkExecutor;
+
+    @Override
+    public void init(AnyValue config) {
+        super.init(config);
+
+        riderWorkExecutor = new ThreadPoolExecutor(
+                8, //corePoolSize 线程池维护线程的最少数量
+                8, // maxPoolSize 线程池维护线程的最大数量
+                60, //keepAliveSeconds 线程池维护线程所允许的空闲时间
+                TimeUnit.SECONDS,
+                new LinkedBlockingDeque<>(Integer.MAX_VALUE),
+                new ThreadFactoryWithNamePrefix("[骑手坐标上传线程池]"));
+    }
+
+    @Override
+    public void destroy(AnyValue config) {
+        super.destroy(config);
+
+        scheduleScanThreadList.forEach(ScheduledThreadPoolExecutor::shutdown);
+
+        if (riderWorkExecutor != null) {
+            riderWorkExecutor.shutdown();
+        }
+    }
+
+    /**
+     * 骑手坐标订单平台需要上传，参考
+     * <a href="https://developer.open-douyin.com/docs/resource/zh-CN/local-life/develop/OpenAPI/takeout/takeout_order/takeout_order_distribution_sync">...</a>
+     * 工作原理
+     * 1、出于订单数据和实时性考量，在订单 骑手已取货 状态 开始后，将订单信息写入到redis的hash中，redis的key由订单id % 4，
+     * 2、hash中的key为订单id，值为 入口平台id，配送平台id，lat，lng
+     * 3、四个线程每个30秒扫各自的hash表，循环所有订单，扔到另一个线程池中处理
+     * 4、在骑手完成后，找到对应的订单，从redis中删除即可
+     *
+     * @param config 参数
+     */
+    @Override
+    public void start(AnyValue config) {
+        super.start(config);
+    }
+
+    public void initScanThread(String dataSourceId, long supplierCode) {
+        IntStream.rangeClosed(0, riderGeoShards).forEach((x) -> {
+            ScheduledThreadPoolExecutor scanThread = new ScheduledThreadPoolExecutor(1, new ThreadFactoryWithNamePrefix("[骑手坐标列表轮询线程_" + x + "_" + supplierCode + "]"));
+            scanThread.scheduleWithFixedDelay(() -> {
+                Map<String, String> orderMap = redisService.hgetAll(keysService.getRedisKey(RedisKeys.KEY_ERP_ORDER_RIDER_GEO, supplierCode, true) + x);
+
+                orderMap.values().forEach((v) -> CompletableFuture.runAsync(() -> {
+                    RiderGeoInfo geoInfo = jsonConvert.convertFromO(RiderGeoInfo.class, v);
+                    String oldLat = geoInfo.getRiderLat();
+                    String oldLng = geoInfo.getRiderLng();
+                    PlatformInfo outPlatform = platformService.getPlatformInfo(geoInfo.getOutPlatformId(), false, supplierCode);
+                    if (outPlatform == null) {
+                        return;
+                    }
+                    PlatformInfo inPlatform = platformService.getPlatformInfo(geoInfo.getInPlatformId(), false, supplierCode);
+                    if (inPlatform == null) {
+                        return;
+                    }
+                    InvokeCallParams geoCallParams = InvokeCallParams.newBuilder()
+                            .businessMethod("Express_RiderLocation_" + outPlatform.getPlatformCode())
+                            .params(jsonConvert.convertTo(geoInfo))
+                            .build();
+                    //Express_RiderLocation接口要求给平台脚本返回的json字符串类型要统一，统一为RiderGeoInfo对象字符串即可
+                    RetResult<InvokeCallResult> geoResult = expressOutService.callExpress(geoCallParams, ERPUtils.getSysTokenUser(), dataSourceId, supplierCode);
+                    if (geoResult.isSuccess()) {
+                        //todo 判断位置相同，其实在脚本里面判断应该就可以了，免得两次json操作
+                        RiderGeoInfo newGeoInfo = jsonConvert.convertFromO(RiderGeoInfo.class, geoResult.getResult().getData());
+                        if ((!StringUtils.equalsIgnoreCase(oldLat, newGeoInfo.getRiderLat())) || (!StringUtils.equalsIgnoreCase(oldLng, newGeoInfo.getRiderLng()))) {
+                            //上传订单平台
+                            InvokeCallParams syncCallParams = InvokeCallParams.newBuilder()
+                                    .businessMethod("Order_Sync_RiderLocation" + inPlatform.getPlatformCode())
+                                    .params(geoResult.getResult().getData())
+                                    .build();
+                            RetResult<InvokeCallResult> syncResult = expressOutService.callExpress(syncCallParams, ERPUtils.getSysTokenUser(), dataSourceId, supplierCode);
+                            if (!syncResult.isSuccess()) {
+                                logger.error("订单同步骑手位置出错：" + inPlatform.getPlatformName() + " " + syncResult.getRetinfo());
+                            } else {
+                                //反写回redis
+                                addOrderGeoInfo(newGeoInfo, supplierCode);
+                            }
+                        }
+
+                    }
+                }, riderWorkExecutor));
+
+
+            }, 30, 40, TimeUnit.SECONDS);
+            scheduleScanThreadList.add(scanThread);
+        });
+    }
+
+    /**
+     * 获取物料对应的订单redis键,分片存储
+     *
+     * @param idOrder      订单id
+     * @param supplierCode 分表
+     * @return redis key
+     */
+    private String getRedisOrderGeoKey(long idOrder, long supplierCode) {
+        return keysService.getRedisKey(RedisKeys.KEY_ERP_ORDER_RIDER_GEO, supplierCode, true) + (idOrder % riderGeoShards);
+    }
+
+    /**
+     * 增加订单扫描位置信息
+     *
+     * @param riderGeoInfo 订单信息
+     * @param supplierCode 分表
+     */
+    public void addOrderGeoInfo(RiderGeoInfo riderGeoInfo, long supplierCode) {
+        String key = getRedisOrderGeoKey(riderGeoInfo.getIdOrder(), supplierCode);
+        redisService.hset(key, String.valueOf(riderGeoInfo.getIdOrder()), jsonConvert.convertTo(riderGeoInfo));
+    }
+
+    /**
+     * 删除订单扫描位置信息
+     *
+     * @param idOrder      订单id
+     * @param supplierCode 分表
+     */
+    public void delOrderGeoInfo(long idOrder, long supplierCode) {
+        String key = getRedisOrderGeoKey(idOrder, supplierCode);
+        redisService.hdel(key, String.valueOf(idOrder));
+    }
+}

ddCommon/src/main/java/com/dderp/common/api/flycat/ExpressGeoService.java

@@ -0,0 +1,23 @@
+package com.dderp.common.api.flycat;
+
+import com.dderp.common.api.ScriptService;
+import com.dderp.common.entity.geo.RiderGeoInfo;
+
+public interface ExpressGeoService extends ScriptService {
+
+    /**
+     * 增加订单扫描位置信息
+     *
+     * @param riderGeoInfo 订单信息
+     * @param supplierCode 分表
+     */
+    void addOrderGeoInfo(RiderGeoInfo riderGeoInfo, long supplierCode);
+
+    /**
+     * 删除订单扫描位置信息
+     *
+     * @param idOrder      订单id
+     * @param supplierCode 分表
+     */
+    void delOrderGeoInfo(long idOrder, long supplierCode);
+}

ddCommon/src/main/java/com/dderp/common/datas/RedisKeys.java

@@ -335,6 +335,11 @@ public class RedisKeys {
      */
     public static final String KEY_DOUYIN_CALL_MSG_ID = "redis.deliver.douyin.call.msgid";
 
+    /**
+     * 订单骑手坐标
+     */
+    public static final String KEY_ERP_ORDER_RIDER_GEO = "redis.erp.order.rider.geo";
+
     private RedisKeys() {
     }
 }

ddCommon/src/main/java/com/dderp/common/entity/geo/RiderGeoInfo.java

@@ -0,0 +1,181 @@
+package com.dderp.common.entity.geo;
+
+import com.dderp.common.entity.base.BaseEntity;
+
+public class RiderGeoInfo extends BaseEntity {
+
+    private long idOrder;
+
+    private long inPlatformId;
+
+    private long outPlatformId;
+
+    private String riderName;
+
+    private String riderPhone;
+
+    private String riderLng;
+
+    private String riderLat;
+
+    private String dataSourceId;
+
+    private long supplierCode;
+
+    public RiderGeoInfo() {
+    }
+
+
+    private RiderGeoInfo(Builder builder) {
+        setIdOrder(builder.idOrder);
+        setInPlatformId(builder.inPlatformId);
+        setOutPlatformId(builder.outPlatformId);
+        setRiderName(builder.riderName);
+        setRiderPhone(builder.riderPhone);
+        setRiderLng(builder.riderLng);
+        setRiderLat(builder.riderLat);
+        setDataSourceId(builder.dataSourceId);
+        setSupplierCode(builder.supplierCode);
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+
+    public long getIdOrder() {
+        return idOrder;
+    }
+
+    public void setIdOrder(long idOrder) {
+        this.idOrder = idOrder;
+    }
+
+    public long getInPlatformId() {
+        return inPlatformId;
+    }
+
+    public void setInPlatformId(long inPlatformId) {
+        this.inPlatformId = inPlatformId;
+    }
+
+    public long getOutPlatformId() {
+        return outPlatformId;
+    }
+
+    public void setOutPlatformId(long outPlatformId) {
+        this.outPlatformId = outPlatformId;
+    }
+
+    public String getRiderName() {
+        return riderName;
+    }
+
+    public void setRiderName(String riderName) {
+        this.riderName = riderName;
+    }
+
+    public String getRiderPhone() {
+        return riderPhone;
+    }
+
+    public void setRiderPhone(String riderPhone) {
+        this.riderPhone = riderPhone;
+    }
+
+    public String getRiderLng() {
+        return riderLng;
+    }
+
+    public void setRiderLng(String riderLng) {
+        this.riderLng = riderLng;
+    }
+
+    public String getRiderLat() {
+        return riderLat;
+    }
+
+    public void setRiderLat(String riderLat) {
+        this.riderLat = riderLat;
+    }
+
+    public String getDataSourceId() {
+        return dataSourceId;
+    }
+
+    public void setDataSourceId(String dataSourceId) {
+        this.dataSourceId = dataSourceId;
+    }
+
+    public long getSupplierCode() {
+        return supplierCode;
+    }
+
+    public void setSupplierCode(long supplierCode) {
+        this.supplierCode = supplierCode;
+    }
+
+    public static final class Builder {
+        private long idOrder;
+        private long inPlatformId;
+        private long outPlatformId;
+        private String riderName;
+        private String riderPhone;
+        private String riderLng;
+        private String riderLat;
+        private String dataSourceId;
+        private long supplierCode;
+
+        private Builder() {
+        }
+
+        public Builder idOrder(long val) {
+            idOrder = val;
+            return this;
+        }
+
+        public Builder inPlatformId(long val) {
+            inPlatformId = val;
+            return this;
+        }
+
+        public Builder outPlatformId(long val) {
+            outPlatformId = val;
+            return this;
+        }
+
+        public Builder riderName(String val) {
+            riderName = val;
+            return this;
+        }
+
+        public Builder riderPhone(String val) {
+            riderPhone = val;
+            return this;
+        }
+
+        public Builder riderLng(String val) {
+            riderLng = val;
+            return this;
+        }
+
+        public Builder riderLat(String val) {
+            riderLat = val;
+            return this;
+        }
+
+        public Builder dataSourceId(String val) {
+            dataSourceId = val;
+            return this;
+        }
+
+        public Builder supplierCode(long val) {
+            supplierCode = val;
+            return this;
+        }
+
+        public RiderGeoInfo build() {
+            return new RiderGeoInfo(this);
+        }
+    }
+}

ddWebCore/src/main/java/com/dderp/webcore/rest/PlatformRest.java

@@ -50,7 +50,7 @@ public class PlatformRest extends BaseService {
     @RestMapping(name = "getPlatformInfo", auth = true, sort = 2, comment = "获取平台档案", methods = {"GET", "POST"})
     @WebApiBean(result = true, type = PlatformInfo.class)
     public CompletableFuture<RMap> getPlatformInfo(
-            @RestParam(name = "idPlatform", comment = "平台档案id") long idPlatform,
+            @RestParam(name = "id", comment = "平台档案id") long idPlatform,
             @RestParam(name = "withDetail", comment = "是否返回属性明细") boolean withDetail,
             @RestHeader(name = ERPHeader.HTTPHEADER_DATASOURCE) String dataSourceId,
             @RestHeader(name = ERPHeader.HTTPHEADER_SUPPLIER) String supplierCode) {