任务提交 流水

common/fire-common/src/main/java/com/fire/common/constants/RocketTopic.java

@@ -18,6 +18,10 @@ public class RocketTopic {
      * 缓存更新topic
      */
     public static final String UPDATE_TOPIC = "update";
+    /**
+     * 流水topic
+     */
+    public static final String TRANSACTION_TOPIC = "transaction";
     /**
      * 无法找到
      */

common/fire-dto/src/main/java/com/fire/dto/TransactionFlow.java

@@ -0,0 +1,56 @@
+package com.fire.dto;
+
+import java.util.Date;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@ApiModel(value = "交易流水表实体")
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TransactionFlow {
+
+    @ApiModelProperty(value = "流水号")
+    private Long seqNo;
+
+    @ApiModelProperty(value = "入账时间")
+    private Date createTime;
+
+    @ApiModelProperty(value = "服务类型:1、订单扣款 2、订单退款 3、财务加款4、财务撤销")
+    private Integer serviceType;
+
+    @ApiModelProperty(value = "操作前金额")
+    private Long beforeAmount;
+
+    @ApiModelProperty(value = "操作金额")
+    private Long operatingAmount;
+
+    @ApiModelProperty(value = "操作后金额")
+    private Long afterAmount;
+
+    @ApiModelProperty(value = "1、客户流水；2、供应商流水")
+    private Integer distinguish;
+
+    @ApiModelProperty(value = "客户/供应商账户")
+    private Long relationId;
+
+    @ApiModelProperty(value = "商户订单号")
+    private String extorderId;
+
+    @ApiModelProperty(value = "业务描述")
+    private String note;
+
+    @ApiModelProperty(value = "订单号")
+    private Long orderId;
+
+    @ApiModelProperty(value = "操作人id")
+    private Long operatorId;
+
+
+}

common/fire-dto/src/main/java/com/fire/dto/enums/RelationType.java

@@ -0,0 +1,25 @@
+package com.fire.dto.enums;
+
+public enum RelationType {
+
+    RELATION_CUSTOMER(1, "客户"),
+    RELATION_SUPPLIER(2, "供应商");
+
+    private final Integer type;
+
+    private final String message;
+
+    RelationType(Integer type, String message) {
+        this.type = type;
+        this.message = message;
+    }
+
+    public Integer type() {
+        return type;
+    }
+
+    public String message() {
+        return message;
+    }
+
+}

common/fire-dto/src/main/java/com/fire/dto/enums/ServiceType.java

@@ -0,0 +1,27 @@
+package com.fire.dto.enums;
+
+public enum ServiceType {
+
+    SERVICE_ORDER_SUB(1, "订单扣款"),
+    SERVICE_ORDER_ADD(2, "订单退款"),
+    SERVICE_FINANCE_ADD(3, "财务加款"),
+    SERVICE_FINANCE_SUB(4, "财务撤销");
+
+    private final Integer type;
+
+    private final String message;
+
+    ServiceType(Integer type, String message) {
+        this.type = type;
+        this.message = message;
+    }
+
+    public Integer type() {
+        return type;
+    }
+
+    public String message() {
+        return message;
+    }
+
+}

common/fire-utils/src/main/java/com/fire/utils/date/DateUtils.java

@@ -28,6 +28,17 @@ public class DateUtils {
             return 0L;
         }
     }
+
+    /**
+     * 老系统时间戳获取
+     *
+
+     * @return String
+     */
+    public static String oldTimestamp() {
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmssSSS");
+        return sdf.format(new Date());
+    }
     
     /**
     * @Description:  TODO  当前系统时间转换

modules/callback-customer/pom.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>modules</artifactId>
+        <groupId>com.blue</groupId>
+        <version>1.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>callback-customer</artifactId>
+
+    <properties>
+        <maven.compiler.source>16</maven.compiler.source>
+        <maven.compiler.target>16</maven.compiler.target>
+    </properties>
+
+</project>

modules/callback-customer/src/main/java/com/fire/customer/callback/CallbackCustomerApplication.java

@@ -0,0 +1,27 @@
+package com.fire.customer.callback;
+
+import com.spring4all.swagger.EnableSwagger2Doc;
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * 回调客户
+ *
+ * @author ZJQ 2021年6月23日16:17:30
+ * @apiNote 此项目只对接一个接口
+ */
+@SpringBootApplication(scanBasePackages = {"com.fire.customer.callback", "com.fire.common.http", "com.fire.common.redis", "com.fire.common.auth"})
+@EnableDiscoveryClient
+@EnableSwagger2Doc
+@MapperScan({"com.fire.customer.callback.mapper"})
+@EnableScheduling
+@EnableAsync
+public class CallbackCustomerApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(CallbackCustomerApplication.class, args);
+    }
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/consumer/RocketOrderConsumer.java

@@ -0,0 +1,68 @@
+package com.fire.customer.callback.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fire.common.exception.BaseException;
+import com.fire.customer.callback.service.CallbackCustomerService;
+import com.fire.dto.FlowOrderInfo;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+import static com.fire.common.constants.RocketTags.FAIL_TAG;
+import static com.fire.common.constants.RocketTags.SUCCESS_TAG;
+import static com.fire.common.constants.RocketTopic.ORDER_TOPIC;
+
+
+/**
+ * 消息消费者，可与生产者分离
+ * 这里定义的messageMode是集群消费，如果是同一消费组则会组内消费者均衡消费；
+ * 如果是不同消费组，则会起到广播消费的效果
+ * <p>
+ * 这里消费成功和失败的订单
+ *
+ * @author ZJQ 2021年6月23日17:01:53
+ */
+
+@Slf4j
+@Component
+@RefreshScope
+@RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}", topic = ORDER_TOPIC, consumeThreadMax = 100, selectorExpression = FAIL_TAG + "||" + SUCCESS_TAG)
+public class RocketOrderConsumer implements RocketMQListener<MessageExt> {
+
+    @Value("${sleep.consume:false}")
+    private Boolean sleepConsume;
+    @Resource
+    private CallbackCustomerService callbackCustomerService;
+
+    @Override
+    public void onMessage(MessageExt msg) {
+
+        if (sleepConsume) {
+            try {
+                Thread.sleep(60000);
+            } catch (InterruptedException e) {
+                log.info("线程sleep失败");
+            }
+            throw new BaseException("暂停消费");
+        }
+        String orderStr = new String(msg.getBody());
+        ObjectMapper om = new ObjectMapper();
+
+        try {
+            FlowOrderInfo orderInfo = om.readValue(orderStr, FlowOrderInfo.class);
+            callbackCustomerService.callbackCustomer(orderInfo);
+        } catch (Exception e) {
+            log.error("订单消息解析失败，消息体为：" + orderStr, e);
+            throw new BaseException("订单消息解析失败", e);
+        }
+
+
+    }
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/data/DataPool.java

@@ -0,0 +1,22 @@
+package com.fire.customer.callback.data;
+
+import com.fire.dto.FlowAppInfo;
+import com.fire.dto.OperatorMaintenance;
+import com.fire.dto.PhoneZone;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 需要在内存中缓存的信息
+ *
+ * @author ZJQ 2021年5月17日16:14:47
+ */
+public class DataPool {
+
+    /**
+     * 客户信息
+     */
+    public static Map<String, FlowAppInfo> flowAppInfoMap = new HashMap<>();
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/dto/CallbackRequestContent.java

@@ -0,0 +1,38 @@
+package com.fire.customer.callback.dto;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+@ApiModel(value = "请求参数正文")
+@JsonIgnoreProperties(ignoreUnknown = true)
+@Data
+public class CallbackRequestContent {
+
+    @ApiModelProperty(value = "客户订单")
+    @JsonProperty(value = "EXTORDER")
+    private String extOrder;
+
+    @ApiModelProperty(value = "订单号")
+    @JsonProperty(value = "ORDERID")
+    private Long orderId;
+
+    @ApiModelProperty(value = "状态")
+    @JsonProperty(value = "STATUS")
+    private String status;
+
+    @ApiModelProperty(value = "状态码")
+    @JsonProperty(value = "CODE")
+    private String code;
+
+    @ApiModelProperty(value = "校验时间")
+    @JsonProperty(value = "CHECKTIME")
+    private String checkTime;
+
+    @ApiModelProperty(value = "结算价格")
+    @JsonProperty(value = "PRICE")
+    private Long price;
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/dto/CallbackRequestMsgBody.java

@@ -0,0 +1,17 @@
+package com.fire.customer.callback.dto;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+@ApiModel(value = "请求参数消息体")
+@JsonIgnoreProperties(ignoreUnknown = true)
+@Data
+public class CallbackRequestMsgBody {
+
+    @ApiModelProperty(value = "请求正文")
+    @JsonProperty(value = "CONTENT")
+    private CallbackRequestContent content;
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/dto/CallbackRequestParam.java

@@ -0,0 +1,23 @@
+package com.fire.customer.callback.dto;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fire.param.HeaderDto;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+@ApiModel(value = "下单接口请求参数")
+@JsonIgnoreProperties(ignoreUnknown = true)
+@Data
+public class CallbackRequestParam {
+
+    @ApiModelProperty(value = "请求参数头信息")
+    @JsonProperty(value = "HEADER")
+    private HeaderDto header;
+
+    @ApiModelProperty(value = "请求参数消息体")
+    @JsonProperty(value = "MSGBODY")
+    private CallbackRequestMsgBody msgBody;
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/mapper/FlowAppInfoMapper.java

@@ -0,0 +1,24 @@
+package com.fire.customer.callback.mapper;
+
+import com.fire.dto.FlowAppInfo;
+
+import java.util.List;
+
+/**
+ * 客户接入表(FlowAppInfo)表数据库访问层
+ *
+ * @author ZJQ  2021-05-17 14:08:29
+ */
+public interface FlowAppInfoMapper {
+
+    /**
+     * 查询所有接入信息
+     *
+     * @return 实例对象list
+     */
+    List<FlowAppInfo> queryAll();
+
+
+
+}
+

modules/callback-customer/src/main/java/com/fire/customer/callback/service/CacheService.java

@@ -0,0 +1,17 @@
+package com.fire.customer.callback.service;
+
+
+/**
+ * 客户服务层
+ *
+ * @author ZJQ 2021年5月17日15:51:14
+ */
+public interface CacheService {
+
+    /**
+     * 缓存客户信息
+     */
+    void cacheCustomer();
+
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/service/CallbackCustomerService.java

@@ -0,0 +1,18 @@
+package com.fire.customer.callback.service;
+
+
+import com.fire.dto.FlowOrderInfo;
+
+/**
+ * 下单服务接口
+ *
+ * @author ZJQ 2021年5月14日17:13:54
+ */
+public interface CallbackCustomerService {
+
+    /**
+     * 回调方法
+     */
+    void callbackCustomer(FlowOrderInfo orderInfo);
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/service/impl/CacheServiceImpl.java

@@ -0,0 +1,45 @@
+package com.fire.customer.callback.service.impl;
+
+
+import com.alibaba.nacos.common.utils.CollectionUtils;
+import com.fire.customer.callback.data.DataPool;
+import com.fire.customer.callback.mapper.FlowAppInfoMapper;
+import com.fire.customer.callback.service.CacheService;
+import com.fire.dto.FlowAppInfo;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 客户服务实现层
+ *
+ * @author ZJQ 2021年5月14日17:13:54
+ */
+@Service
+public class CacheServiceImpl implements CacheService {
+
+    @Resource
+    private FlowAppInfoMapper flowAppInfoMapper;
+
+
+    /**
+     * 缓存客户信息
+     */
+    @Override
+    public void cacheCustomer() {
+        //查询所有的客户信息和所有的接入信息
+        List<FlowAppInfo> appInfos = flowAppInfoMapper.queryAll();
+
+        if (CollectionUtils.isNotEmpty(appInfos)) {
+
+            //将接入信息list 转换为appId为key的map 如果存在重复key，以后出现的为准
+            DataPool.flowAppInfoMap = appInfos.stream().collect(Collectors.toMap(FlowAppInfo::getAppId, a -> a, (k1, k2) -> k2));
+        }
+
+
+    }
+
+
+}

modules/callback-customer/src/main/java/com/fire/customer/callback/service/impl/CallbackCustomerServiceImpl.java

@@ -0,0 +1,69 @@
+package com.fire.customer.callback.service.impl;
+
+
+import com.fire.customer.callback.data.DataPool;
+import com.fire.customer.callback.dto.CallbackRequestContent;
+import com.fire.customer.callback.dto.CallbackRequestMsgBody;
+import com.fire.customer.callback.dto.CallbackRequestParam;
+import com.fire.customer.callback.service.CallbackCustomerService;
+import com.fire.dto.FlowAppInfo;
+import com.fire.dto.FlowOrderInfo;
+import com.fire.param.HeaderDto;
+import com.fire.utils.date.DateUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.util.DigestUtils;
+import org.springframework.web.client.RestTemplate;
+
+import javax.annotation.Resource;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * 向客户回调
+ *
+ * @author ZJQ 2021年6月23日16:37:01
+ */
+@Service
+@Slf4j
+public class CallbackCustomerServiceImpl implements CallbackCustomerService {
+
+    @Resource
+    private RestTemplate restTemplate;
+
+    @Override
+    public void callbackCustomer(FlowOrderInfo orderInfo) {
+        //创建回调参数
+        CallbackRequestParam requestParam = new CallbackRequestParam();
+        //参数header
+        HeaderDto header = new HeaderDto();
+        //参数body
+        CallbackRequestMsgBody body = new CallbackRequestMsgBody();
+        //参数内容
+        CallbackRequestContent content = new CallbackRequestContent();
+        body.setContent(content);
+        requestParam.setHeader(header);
+        requestParam.setMsgBody(body);
+        //获取老系统的时间戳
+        String oldTimestamp = DateUtils.oldTimestamp();
+        //设置参数内容中的各个属性
+        content.setExtOrder(orderInfo.getExtorderId());
+        content.setCheckTime(oldTimestamp);
+        content.setOrderId(orderInfo.getOrderId());
+        content.setPrice(orderInfo.getPrice());
+        //设置header中的参数
+        header.setTimestamp(oldTimestamp);
+        header.setAppId(orderInfo.getAppId());
+        //待生成
+        header.setSeqNo("CP" + System.currentTimeMillis());
+        header.setVersion("V1.0");
+        //获取客户的appKey
+        FlowAppInfo flowAppInfo = DataPool.flowAppInfoMap.get(header.getAppId());
+        //计算secretKey
+        String secretStr = header.getTimestamp() + header.getSeqNo() + header.getAppId() + flowAppInfo.getAppKey();
+        String secretKey = DigestUtils.md5DigestAsHex(secretStr.getBytes(StandardCharsets.UTF_8));
+        header.setSecretKey(secretKey);
+
+        restTemplate.postForObject(flowAppInfo.getCallbackUrl(), requestParam, String.class);
+
+    }
+}

modules/callback-customer/src/main/resources/bootstrap.yml

@@ -0,0 +1,38 @@
+server:
+  port: 0
+spring:
+  application:
+    name: callback-customer
+  profiles:
+    active: dev
+---
+spring:
+  config:
+    activate:
+      on-profile: dev
+  cloud:
+    nacos:
+      config:
+        server-addr: 47.106.133.48:8848
+        file-extension: yaml
+        namespace: fire
+      discovery:
+        server-addr: 47.106.133.48:8848
+        namespace: fire
+        service: callback-customer
+---
+spring:
+  config:
+    activate:
+      on-profile: test
+  cloud:
+    nacos:
+      config:
+        server-addr: 139.155.226.217:8848
+        file-extension: yaml
+        namespace: fire
+      discovery:
+        server-addr: 139.155.226.217:8848
+        namespace: fire
+        service: callback-customer
+

modules/callback-customer/src/main/resources/logback-spring.xml

@@ -0,0 +1,65 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- 当scan属性设置为true时，配置文件如果发生改变，将会被重新加载，默认值为true。 -->
+<configuration scan="true" scanPeriod="60 seconds" debug="false">
+    <!-- 定义变量值，也可以定义成一个路径，可以使“${}”来使用变量 -->
+    <property name="log_path" value="/data/fire-log/callback-customer"/>
+    <!--<springProperty scope="context" name="modelName" source="spring.application.name"/>-->
+    <property name="log_encoder_code" value="[%d{yyyy-MM-dd HH:mm:ss}] [%thread] [%-5level] [%logger] %msg%n"/>
+    <!-- 设置上下文名称，用于区分不同应用程序的记录，一旦设置，不能修改 -->
+    <contextName>elk_log</contextName>
+    <!-- 获取时间戳字符串 -->
+    <timestamp key="now_time" datePattern="yyyyMMdd'T'HHmmss"/>
+    <!-- 负责写日志的组件，它有两个必要属性name和class。name指定appender名称，class指定appender的全限定名 -->
+    <!--输出到控制台-->
+    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+        <!-- 对日志进行格式化 -->
+        <encoder>
+            <pattern>${log_encoder_code}</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="log_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <!-- 按天回滚 daily -->
+            <fileNamePattern>${log_path}/info-log-%d{yyyy-MM-dd}.log</fileNamePattern>
+            <!-- 日志最大的历史 120天 -->
+            <maxHistory>120</maxHistory>
+        </rollingPolicy>
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+    <appender name="log_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <!-- 过滤器，只记录ERROR级别的日志 -->
+        <filter class="ch.qos.logback.classic.filter.LevelFilter">
+            <level>DEBUG</level>
+            <onMatch>ACCEPT</onMatch>
+            <onMismatch>DENY</onMismatch>
+        </filter>
+        <!-- 最常用的滚动策略，它根据时间来制定滚动策略.既负责滚动也负责出发滚动 -->
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <!--日志输出位置  可相对、和绝对路径 -->
+            <fileNamePattern>${log_path}/error-log-%d{yyyy-MM-dd}.log</fileNamePattern>
+            <!-- 可选节点，控制保留的归档文件的最大数量，超出数量就删除旧文件假设设置每个月滚动，且<maxHistory>是6，
+            则只保存最近6个月的文件，删除之前的旧文件。注意，删除旧文件是，那些为了归档而创建的目录也会被删除-->
+            <maxHistory>120</maxHistory>
+        </rollingPolicy>
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <!-- SQL相关日志输出-->
+<!--    <logger name="org.apache.ibatis" level="INFO" additivity="false"/>-->
+<!--    <logger name="org.mybatis.spring" level="INFO" additivity="false"/>-->
+<!--    <logger name="com.github.miemiedev.mybatis.paginator" level="INFO" additivity="false"/>-->
+    <!-- 把>=DEBUG级别的日志都输出到控制台 -->
+    <root level="INFO">
+        <appender-ref ref="console"/>
+        <!--<appender-ref ref="file" />-->
+        <appender-ref ref="log_info"/>
+        <appender-ref ref="log_error"/>
+    </root>
+    <!-- 屏蔽心跳日志-->
+    <logger name="com.alibaba.nacos" level="error" additivity="false"/>
+</configuration>

modules/callback-customer/src/main/resources/mapper/FlowAppInfoMapper.xml

@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.fire.customer.callback.mapper.FlowAppInfoMapper">
+
+    <resultMap type="com.fire.dto.FlowAppInfo" id="FlowAppInfoMap">
+        <result property="flowAppId" column="flow_app_id" jdbcType="INTEGER"/>
+        <result property="customerId" column="customer_id" jdbcType="INTEGER"/>
+        <result property="appId" column="app_id" jdbcType="VARCHAR"/>
+        <result property="appKey" column="app_key" jdbcType="VARCHAR"/>
+        <result property="startDate" column="start_date" jdbcType="TIMESTAMP"/>
+        <result property="endDate" column="end_date" jdbcType="TIMESTAMP"/>
+        <result property="status" column="status" jdbcType="INTEGER"/>
+        <result property="appName" column="app_name" jdbcType="VARCHAR"/>
+        <result property="callbackUrl" column="callback_url" jdbcType="VARCHAR"/>
+        <result property="ipAddress" column="ip_address" jdbcType="VARCHAR"/>
+        <result property="dispatchChannel" column="dispatch_channel" jdbcType="VARCHAR"/>
+        <result property="totalCount" column="total_count" jdbcType="INTEGER"/>
+        <result property="time" column="time" jdbcType="INTEGER"/>
+    </resultMap>
+
+    <!--查询全部-->
+    <select id="queryAll" resultMap="FlowAppInfoMap">
+        <include refid="baseSql"/>
+    </select>
+
+    <sql id="baseSql">
+        select flow_app_id,
+               customer_id,
+               app_id,
+               app_key,
+               start_date,
+               end_date,
+               status,
+               app_name,
+               callback_url,
+               ip_address,
+               dispatch_channel,
+               total_count
+        from flow_app_info
+    </sql>
+
+</mapper>
+

modules/get-order-supplier/src/main/java/com/fire/get/order/GetOrderSupplierApplication.java

@@ -5,6 +5,7 @@ import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
 /**
@@ -18,6 +19,7 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 @EnableSwagger2Doc
 @MapperScan({"com.fire.get.order.mapper"})
 @EnableScheduling
+@EnableAsync
 public class GetOrderSupplierApplication {
     public static void main(String[] args) {
         SpringApplication.run(GetOrderSupplierApplication.class, args);

modules/get-order-supplier/src/main/java/com/fire/get/order/mapper/MobileFlowDispatchRecMapper.java

@@ -0,0 +1,26 @@
+package com.fire.get.order.mapper;
+
+import com.fire.dto.MobileFlowDispatchRec;
+
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 分发记录(MobileFlowDispatchRec)表数据库访问层
+ *
+ * @author ZJQ  2021-06-01 11:35:23
+ */
+public interface MobileFlowDispatchRecMapper {
+
+    /**
+     * 根据状态和创建时间发送子订单
+     *
+     * @param beforeTime 时间
+     * @param sendStatus 状态
+     * @return 实例对象
+     */
+    List<MobileFlowDispatchRec> queryByStatusTime(Date beforeTime, Integer sendStatus);
+
+
+}
+

modules/get-order-supplier/src/main/java/com/fire/get/order/service/GetOrderSupplierService.java

@@ -2,6 +2,7 @@ package com.fire.get.order.service;
 
 
 import com.fire.dto.FlowOrderInfo;
+import com.fire.dto.MobileFlowDispatchRec;
 
 /**
  * 查询订单
@@ -13,6 +14,6 @@ public interface GetOrderSupplierService {
     /**
      * 下单方法
      */
-    void getOrder(FlowOrderInfo orderInfo);
+    void getOrder(MobileFlowDispatchRec chileOrder);
 
 }

modules/get-order-supplier/src/main/java/com/fire/get/order/service/MobileFlowDispatchRecService.java

@@ -0,0 +1,21 @@
+package com.fire.get.order.service;
+
+import com.fire.dto.MobileFlowDispatchRec;
+
+import java.util.List;
+
+/**
+ * 分发记录(MobileFlowDispatchRec)表数据库访问层
+ *
+ * @author ZJQ  2021-06-01 11:35:23
+ */
+public interface MobileFlowDispatchRecService {
+
+    /**
+     * 查询子订单
+     */
+    void queryDispatchRecs();
+
+
+}
+

modules/get-order-supplier/src/main/java/com/fire/get/order/service/impl/GetOrderSupplierServiceImpl.java

@@ -16,12 +16,15 @@ import com.fire.supplier.dto.*;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestTemplate;
 import redis.clients.jedis.JedisCluster;
 
 import javax.annotation.Resource;
 
+import java.util.Date;
+
 import static com.fire.common.constants.RocketTags.FAIL_TAG;
 import static com.fire.common.constants.RocketTags.SUCCESS_TAG;
 import static com.fire.common.constants.RocketTopic.CHILD_ORDER_TOPIC;
@@ -55,9 +58,10 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
     private SupplierAuth supplierAuth;
 
     @Override
-    public void getOrder(FlowOrderInfo orderInfo) {
+    @Async
+    public void getOrder(MobileFlowDispatchRec childOrder) {
         //根据通道提取供应商信息
-        ChannelSupplier supplier = DataPool.supplierHashMap.get(orderInfo.getChannelId());
+        ChannelSupplier supplier = DataPool.supplierHashMap.get(childOrder.getChannelId());
         //组装请求参数
         QueryOrderRequest queryOrderRequest = new QueryOrderRequest();
         RequestHeader header = new RequestHeader();
@@ -70,7 +74,7 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
         header.setTimestamp((int) (System.currentTimeMillis() / 1000));
         header.setSign("MD5");
         //组装请求参数body 这里发送的是子订单id，规则为主订单id+发送次数sendCount
-        body.setMerchantOrderNo(String.valueOf(orderInfo.getOrderId() + orderInfo.getSendCount()));
+        body.setMerchantOrderNo(String.valueOf(childOrder.getOrderId() + childOrder.getSendCount()));
         //code取值是供应商的passwd 用code和header中的值进行md5加密得到原版本1的 签名
         String code = supplier.getPasswd();
 
@@ -102,18 +106,18 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
         //如果充值中不操作 所以非充值中都需要更改状态
         if (!supplierOrderStatus.equals(SUPPLIER_ORDER_PROCESS)) {
             //分发记录修改 只要不是充值中，子订单就进入终态 todo redis中获取分发记录
-            MobileFlowDispatchRec newChildOrder = orderInfo.getNewChildOrder();
-            newChildOrder.setGwErrorMsg(businessStatus.message());
-            newChildOrder.setGwErrorCode(String.valueOf(businessStatus.status()));
+
+            childOrder.setGwErrorMsg(businessStatus.message());
+            childOrder.setGwErrorCode(String.valueOf(businessStatus.status()));
             switch (supplierOrderStatus) {
-                case SUPPLIER_ORDER_FAILED -> newChildOrder.setSendStatus(ORDER_FAIL.status());
-                case SUPPLIER_ORDER_SUCCESS -> newChildOrder.setSendStatus(ORDER_SUCCESS.status());
+                case SUPPLIER_ORDER_FAILED -> childOrder.setSendStatus(ORDER_FAIL.status());
+                case SUPPLIER_ORDER_SUCCESS -> childOrder.setSendStatus(ORDER_SUCCESS.status());
             }
-            Long recId = newChildOrder.getRecId();
+            Long recId = childOrder.getRecId();
             String childOrderStr;
             ObjectMapper om = new ObjectMapper();
             try {
-                childOrderStr = om.writeValueAsString(newChildOrder);
+                childOrderStr = om.writeValueAsString(childOrder);
                 //订单转换为json更新入redis
                 long hKey = recId / 10000000;
                 jedisCluster.hset(CHILD_ORDER_INFO.key() + hKey, String.valueOf(recId), childOrderStr);
@@ -129,35 +133,37 @@ public class GetOrderSupplierServiceImpl implements GetOrderSupplierService {
                 log.error("子订单在分发的时候转换成json失败：", e);
             }
 
-            String orderStr;
-            try {
-                orderStr = om.writeValueAsString(orderInfo);
-            } catch (Exception e) {
-                log.error("订单重入优先级队列失败：", e);
-                //todo 进入兜底订单
-                return;
-            }
+
+            long orderId = childOrder.getOrderId();
+            //订单转换为json更新入redis
+            long hKey = orderId / 10000000;
+            String orderStr = jedisCluster.hget(ORDER_INFO.key() + hKey, String.valueOf(orderId));
+
             //子订单成功  那么主订单就成功了 子订单成功 那么主订单重入队列
             switch (supplierOrderStatus) {
                 case SUPPLIER_ORDER_SUCCESS -> {
-                    long orderId = orderInfo.getOrderId();
-                    //订单转换为json更新入redis
-                    long hKey = orderId / 10000000;
-                    jedisCluster.hset(ORDER_INFO.key() + hKey, String.valueOf(orderId), orderStr);
-                    //订单入队列
-                    rocketMQTemplate.syncSendOrderly(ORDER_TOPIC + ":" + SUCCESS_TAG, MessageBuilder.withPayload(orderStr).build(), String.valueOf(orderId));
-                }
-                case SUPPLIER_ORDER_FAILED -> {
-                    //入优先级队列
-                    redisPriorityQueueScript.addContent(orderInfo.getOrderId(), orderStr);
+                    try {
+                        FlowOrderInfo orderInfo = om.readValue(orderStr, FlowOrderInfo.class);
+                        orderInfo.setStatus(ORDER_SUCCESS.status());
+                        orderInfo.setChannelId(childOrder.getChannelId());
+                        orderInfo.setNote("成功");
+                        orderInfo.setOperatorBalancePrice(childOrder.getOperatorBalancePrice());
+                        orderInfo.setCallbackTime(new Date());
+                        orderStr = om.writeValueAsString(orderInfo);
+                        jedisCluster.hset(ORDER_INFO.key() + hKey, String.valueOf(orderId), orderStr);
+                        //订单入队列
+                        rocketMQTemplate.syncSendOrderly(ORDER_TOPIC + ":" + SUCCESS_TAG, MessageBuilder.withPayload(orderStr).build(), String.valueOf(orderId));
+                    } catch (JsonProcessingException e) {
+                        log.error("通道返回成功，但是订单代码异常", e);
+                        //todo 这里是成功异常
+                    }
                 }
+                //子订单失败
+                case SUPPLIER_ORDER_FAILED -> redisPriorityQueueScript.addContent(orderId, orderStr);
             }
-
-
         }
 
 
-
     }
 
 }

modules/get-order-supplier/src/main/java/com/fire/get/order/service/impl/MobileFlowDispatchRecServiceImpl.java

@@ -0,0 +1,44 @@
+package com.fire.get.order.service.impl;
+
+
+import com.alibaba.nacos.common.utils.CollectionUtils;
+import com.fire.dto.MobileFlowDispatchRec;
+import com.fire.get.order.mapper.MobileFlowDispatchRecMapper;
+import com.fire.get.order.service.GetOrderSupplierService;
+import com.fire.get.order.service.MobileFlowDispatchRecService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.Calendar;
+import java.util.List;
+
+import static com.fire.dto.enums.OrderStatus.ORDER_SENT;
+
+/**
+ * 查询子订单
+ *
+ * @author ZJQ 2021年6月30日16:37:01
+ */
+@Service
+@Slf4j
+public class MobileFlowDispatchRecServiceImpl implements MobileFlowDispatchRecService {
+
+    @Resource
+    private MobileFlowDispatchRecMapper mobileFlowDispatchRecMapper;
+    @Resource
+    private GetOrderSupplierService getOrderSupplierService;
+
+    @Override
+    public void queryDispatchRecs() {
+        //两分钟前的订单
+        Calendar calendar = Calendar.getInstance();
+        calendar.add(Calendar.MINUTE, -2);
+        List<MobileFlowDispatchRec> recs = mobileFlowDispatchRecMapper.queryByStatusTime(calendar.getTime(), ORDER_SENT.status());
+        //查询订单
+        if (CollectionUtils.isNotEmpty(recs)) {
+            recs.forEach(rec -> getOrderSupplierService.getOrder(rec));
+        }
+
+    }
+}

modules/get-order-supplier/src/main/java/com/fire/get/order/task/QueryChildTask.java

@@ -0,0 +1,32 @@
+package com.fire.get.order.task;
+
+import com.fire.get.order.service.MobileFlowDispatchRecService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 缓存任务
+ *
+ * @author ZJQ 2021年5月31日13:41:01
+ */
+@Component
+@Slf4j
+public class QueryChildTask {
+
+    @Resource
+    private MobileFlowDispatchRecService mobileFlowDispatchRecService;
+
+    /**
+     * 每隔两分钟执行
+     */
+    @Async
+    @Scheduled(cron = "5 1/2 * * * ?")
+    public void cacheCustomer() {
+        mobileFlowDispatchRecService.queryDispatchRecs();
+    }
+
+}

modules/get-order-supplier/src/main/java/com/fire/get/order/test/gushiwen.java

@@ -8,6 +8,8 @@ import org.jsoup.nodes.Element;
 import org.jsoup.select.Elements;
 
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
 public class gushiwen {
 
@@ -20,6 +22,9 @@ public class gushiwen {
             Elements elements = element.select("a");
             for (Element element1 : elements) {
                 String type = element1.text();
+                if (!type.equals("送别")) {
+                    continue;
+                }
                 String ur = yuming + element1.attr("href");
                 try {
                     Connection connect1 = Jsoup.connect(ur);
@@ -40,9 +45,18 @@ public class gushiwen {
                             Connection detailConn = Jsoup.connect(detailUrl);
                             Document detailAllDoc = detailConn.get();
                             Element elementsDetail = detailAllDoc.getElementsByClass("sons").get(0);
-                            String contEle = elementsDetail.getElementsByClass("contson").get(0).html();
-                            System.out.println(type + "    " + type1 + "    " + name + "    " + auth + "    " + detailUrl);
-                            System.out.println(elementsDetail.toString());
+                            String cont = elementsDetail.getElementsByClass("contson").get(0).html();
+
+                            Elements others = detailAllDoc.getElementsByClass("contyishang");
+
+                            Map<String,String> otherMap = new HashMap<>();
+
+                            for (Element other : others) {
+
+                            }
+
+                            System.out.println(type + "    " + type1 + "    " + name + "    " + auth + "    " + cont);
+
                             break;
                         }
                         break;

modules/get-order-supplier/src/main/resources/mapper/MobileFlowDispatchRecMapper.xml

@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.fire.get.order.mapper.MobileFlowDispatchRecMapper">
+
+    <resultMap type="com.fire.dto.MobileFlowDispatchRec" id="MobileFlowDispatchRecMap">
+        <result property="recId" column="rec_id" jdbcType="VARCHAR"/>
+        <result property="orderId" column="order_id" jdbcType="VARCHAR"/>
+        <result property="phoneNo" column="phone_no" jdbcType="VARCHAR"/>
+        <result property="packageId" column="package_id" jdbcType="VARCHAR"/>
+        <result property="flowAmount" column="flow_amount" jdbcType="INTEGER"/>
+        <result property="sendCount" column="send_count" jdbcType="INTEGER"/>
+        <result property="createDate" column="create_date" jdbcType="TIMESTAMP"/>
+        <result property="callbackTime" column="callback_time" jdbcType="TIMESTAMP"/>
+        <result property="sendStatus" column="send_status" jdbcType="INTEGER"/>
+        <result property="gwSeqNo" column="gw_seq_no" jdbcType="VARCHAR"/>
+        <result property="gwStatus" column="gw_status" jdbcType="VARCHAR"/>
+        <result property="gwErrorCode" column="gw_error_code" jdbcType="VARCHAR"/>
+        <result property="gwErrorMsg" column="gw_error_msg" jdbcType="VARCHAR"/>
+        <result property="phoneHome" column="phone_home" jdbcType="VARCHAR"/>
+        <result property="appId" column="app_id" jdbcType="VARCHAR"/>
+        <result property="lastCallbackMsg" column="last_callback_msg" jdbcType="VARCHAR"/>
+        <result property="operatorBalancePrice" column="operator_balance_price" jdbcType="INTEGER"/>
+        <result property="batchCount" column="batch_count" jdbcType="INTEGER"/>
+        <result property="customerProductsId" column="customer_products_id" jdbcType="VARCHAR"/>
+        <result property="customerName" column="customer_name" jdbcType="VARCHAR"/>
+        <result property="channelProductsId" column="channel_products_id" jdbcType="VARCHAR"/>
+        <result property="operatorId" column="operator_id" jdbcType="VARCHAR"/>
+        <result property="recordLog" column="record_log" jdbcType="VARCHAR"/>
+        <result property="channelName" column="channel_name" jdbcType="VARCHAR"/>
+        <result property="phoneOperator" column="phone_operator" jdbcType="INTEGER"/>
+    </resultMap>
+
+    <!--查询单个-->
+    <select id="queryByStatusTime" resultMap="MobileFlowDispatchRecMap">
+        <include refid="baseSql"/>
+        where create_date &lt; #{beforeTime} and send_status = #{sendStatus}
+    </select>
+
+
+    <sql id="baseSql">
+        select rec_id,
+               order_id,
+               phone_no,
+               package_id,
+               flow_amount,
+               send_count,
+               create_date,
+               callback_time,
+               send_status,
+               gw_seq_no,
+               gw_status,
+               gw_error_code,
+               gw_error_msg,
+               phone_home,
+               app_id,
+               last_callback_msg,
+               operator_balance_price,
+               batch_count,
+               customer_products_id,
+               customer_name,
+               channel_products_id,
+               operator_id,
+               record_log,
+               channel_name,
+               phone_operator
+        from mobile_flow_dispatch_rec
+    </sql>
+</mapper>
+

modules/make-order/src/main/java/com/fire/order/service/impl/CacheServiceImpl.java

@@ -99,6 +99,8 @@ public class CacheServiceImpl implements CacheService {
         List<VirtualNum> virtualNums = virtualNumMapper.queryAll();
         if (CollectionUtils.isNotEmpty(virtualNums)) {
             DataPool.virtualMap = virtualNums.stream().collect(Collectors.toMap(VirtualNum::getVirtualNum, a -> Optional.ofNullable(a.getNote()).orElse("虚拟号段"), (k1, k2) -> k2));
+        } else {
+            DataPool.virtualMap = new HashMap<>();
         }
 
     }
@@ -128,6 +130,8 @@ public class CacheServiceImpl implements CacheService {
                 maintenanceMap.put(key, om);
             }
             DataPool.maintenanceMap = maintenanceMap;
+        } else {
+            DataPool.maintenanceMap = new HashMap<>();
         }
     }
 }

modules/make-order/src/main/java/com/fire/order/service/impl/MakeOrderServiceImpl.java

@@ -41,6 +41,8 @@ import static com.fire.dto.enums.OrderStatus.ORDER_IN_QUEUE;
 import static com.fire.dto.enums.PriceCheck.CHECK;
 import static com.fire.dto.enums.Province.QG_ALL;
 import static com.fire.dto.enums.RedisKey.*;
+import static com.fire.dto.enums.RelationType.RELATION_CUSTOMER;
+import static com.fire.dto.enums.ServiceType.SERVICE_ORDER_SUB;
 import static com.fire.dto.enums.Status.*;
 import static com.fire.dto.enums.ValidStatus.INVALID;
 
@@ -355,6 +357,22 @@ public class MakeOrderServiceImpl implements MakeOrderService {
             return responseDto;
         }
 
+        //自增id 取自redis，长度16位，
+        Long transSeq = jedisCluster.incr(GLOBAL_ID_INCR.key());
+        TransactionFlow transaction = TransactionFlow.builder()
+                .afterAmount(afterAmount)
+                .beforeAmount(afterAmount + product.getPrice())
+                .createTime(new Date())
+                .distinguish(RELATION_CUSTOMER.type())
+                .extorderId(extOrder)
+                .orderId(orderId)
+                .note("订单充值扣减")
+                .operatingAmount(product.getPrice())
+                .serviceType(SERVICE_ORDER_SUB.type())
+                .relationId(flowAppInfo.getCustomerId())
+                .seqNo(transSeq)
+                .build();
+
         try {
             //订单入redis
             long hKey = orderId / 10000000;
@@ -363,7 +381,8 @@ public class MakeOrderServiceImpl implements MakeOrderService {
             //订单入队列
             rocketMQTemplate.syncSendOrderly(ORDER_TOPIC + ":" + MAKE_TAG, MessageBuilder.withPayload(orderStr).build(), String.valueOf(orderId));
             long time6 = System.currentTimeMillis();
-            log.info("取id" + (time2 - time1) + " 校验id" + (time3 - time2) + " 扣减" + (time4 - time3) + " 订单入redis" + (time5 - time4) + " 消息入队列" + (time6 - time5)+" 总"+(time6-time1));
+            log.info("取id" + (time2 - time1) + " 校验id" + (time3 - time2) + " 扣减" + (time4 - time3) + " 订单入redis" + (time5 - time4) + " 消息入队列" + (time6 - time5) + " 总" + (time6 - time1));
+
         } catch (Exception e) {
             //其他异常 扣减补偿
             log.error("订单存储或者发送到队列失败", e);

modules/pom.xml

@@ -20,6 +20,7 @@
         <module>get-order</module>
         <module>make-order-supplier</module>
         <module>get-order-supplier</module>
+        <module>callback-customer</module>
     </modules>
 
     <dependencies>