提交支付查询

liquidnet-bus-api/liquidnet-service-dragon-api/src/main/java/com/liquidnet/service/dragon/service/IDragonOrdersService.java

@@ -8,8 +8,6 @@ import com.liquidnet.service.dragon.dto.DragonPayOrderQueryRespDto;
 import javax.servlet.http.HttpServletRequest;
 
 public interface IDragonOrdersService {
-    void sendRedisQueue();
-
     ResponseDto<DragonPayBaseRespDto> dragonPay(DragonPayBaseReqDto dragonPayBaseReqDto);
 
     void dragonNotify(HttpServletRequest request,String payType,String deviceFrom);

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-dragon/src/main/java/com/liquidnet/service/consumer/dragon/service/receiver/RedisPayReceiver.java

 package com.liquidnet.service.consumer.dragon.service.receiver;
 
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.consumer.dragon.service.IBaseDao;
+import com.liquidnet.service.dragon.constant.DragonConstant;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.stream.StreamListener;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
+import java.util.HashMap;
+
 @Slf4j
 @Component
 public class RedisPayReceiver implements StreamListener<String, MapRecord<String, String, String>> {
 
+    @Resource
+    IBaseDao baseDao;
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
     @Override
     public void onMessage(MapRecord<String, String, String> message) {
-        log.info("接受到来自redis PAY 的消息");
-        System.out.println("message id "+message.getId());
-        System.out.println("stream "+message.getStream());
-        System.out.println("body "+message.getValue());
+        log.info("接受到来自redis REFUND 的消息");
+        System.out.println("message id " + message.getId());
+        System.out.println("stream " + message.getStream());
+        System.out.println("body " + message.getValue());
+        consumerSqlDaoHandler(message.getValue().get("message"));
+    }
+
+    private void consumerSqlDaoHandler(String msg) {
+        try {
+            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
+            log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+            log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                //应答
+            } else {
+                sendMySqlRedis(msg);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 给 REDIS 队列发送消息 数据库相关
+     *
+     * @param msg 接收到的内容
+     * @return
+     */
+    private boolean sendMySqlRedis(String msg) {
+        try {
+            HashMap<String, String> map = new HashMap<>();
+            map.put("message", msg);
+            MapRecord<String, String, String> record = StreamRecords.mapBacked(map).withStreamKey(DragonConstant.MysqlRedisQueueEnum.DRAGON_REFUND_KEY.getCode());
+            stringRedisTemplate.opsForStream().add(record);
+            return true;
+        } catch (Exception e) {
+            e.printStackTrace();
+            return false;
+        }
     }
 }

liquidnet-bus-service/liquidnet-service-dragon/liquidnet-service-dragon-impl/src/main/java/com/liquidnet/service/dragon/service/impl/DragonOrdersServiceImpl.java

@@ -15,13 +15,10 @@ import com.liquidnet.service.dragon.service.IDragonOrdersService;
 import com.liquidnet.service.dragon.utils.DataUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.redis.connection.stream.MapRecord;
-import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
 import javax.servlet.http.HttpServletRequest;
-import java.util.HashMap;
 
 @Slf4j
 @Service
@@ -35,18 +32,6 @@ public class DragonOrdersServiceImpl implements IDragonOrdersService {
     @Autowired
     private DataUtils dataUtils;
 
-    @Override
-    public void sendRedisQueue() {
-        try {
-            HashMap<String ,String> map = new HashMap<>();
-            map.put("message","测试 redis 订阅信息1");
-            MapRecord<String, String, String> record = StreamRecords.mapBacked(map).withStreamKey("dragon-pay");
-            stringRedisTemplate.opsForStream().add(record);
-        }catch (Exception e){
-            e.printStackTrace();
-        }
-    }
-
     @Override
     public ResponseDto<DragonPayBaseRespDto> dragonPay(DragonPayBaseReqDto dragonPayBaseReqDto) {
         //设置支付编号