优化支付代码

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

@@ -30,10 +30,18 @@ public class RedisPayReceiver implements StreamListener<String, MapRecord<String
         System.out.println("message id " + message.getId());
         System.out.println("stream " + message.getStream());
         System.out.println("body " + message.getValue());
-        consumerSqlDaoHandler(message.getValue().get("message"));
+        boolean result = consumerSqlDaoHandler(message.getValue().get("message"));
+
+        // 消费成功确认，消息删除和消息确认是一个事务
+        if(result){
+            stringRedisTemplate.multi();
+            stringRedisTemplate.opsForStream().delete(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_KEY.getCode(), message.getId());
+//            stringRedisTemplate.opsForStream().acknowledge(DragonConstant.MysqlRedisQueueEnum.DRAGON_PAY_GROUP.getCode(), message);
+            stringRedisTemplate.exec();
+        }
     }
 
-    private void consumerSqlDaoHandler(String msg) {
+    private boolean consumerSqlDaoHandler(String msg) {
         try {
             SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
             log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
@@ -41,11 +49,12 @@ public class RedisPayReceiver implements StreamListener<String, MapRecord<String
             Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
             log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
             if (rstBatchSqls) {
-                //应答
+                return true;
             }
         } catch (Exception e) {
             e.printStackTrace();
         }
+        return false;
     }
 
     /**