MQ消费调整;

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/AbstractSmsRedisReceiver.java

 package com.liquidnet.service.consumer.adam.receiver;
 
 import com.liquidnet.common.sms.processor.SmsProcessor;
+import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.SmsMessage;
 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 javax.annotation.Resource;
+import java.util.HashMap;
 
 @Slf4j
 public abstract class AbstractSmsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
@@ -25,29 +28,39 @@ public abstract class AbstractSmsRedisReceiver implements StreamListener<String,
                 this.getRedisStreamKey(), message.getId(), message.getStream(), message.getValue());
 
         boolean result = this.consumerSmsSendHandler(message.getValue().get("message"));
+        log.info("CONSUMER SMS RESULT:{} ==> MESSAGE_ID:{}", result, message.getId());
 
-        log.debug("CONSUMER SMS RESULT:{}", result);
-
-        // 消费成功确认，消息删除和消息确认是一个事务
-        if (result) {
-            log.info("CONSUMER SMS SUCC ==> MESSAGE_ID:{}", message.getId());
+        try {
+            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+            stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
+        } catch (Exception e) {
+            log.error("CONSUMER SMS RESULT:{} ==> DEL_REDIS_QUEUE_MSG_EXCEPTION[MESSAGE_ID:{},MSG:{}]", result, message.getId(), JsonUtils.toJson(message), e);
+        } finally {
             try {
-                stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
-            } catch (Exception e) {
-                log.error("CONSUMER SMS SUC ==> DEL_REDIS_QUEUE_MSG_EXCEPTION[MESSAGE_ID:{}]", message.getId(), e);
+                stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+            } catch (Exception ignored) {
             }
         }
     }
 
     private boolean consumerSmsSendHandler(String msg) {
+        boolean aBoolean = false;
         try {
             SmsMessage smsMessage = JsonUtils.fromJson(msg, SmsMessage.class);
-            return smsProcessor.send(smsMessage.getPhone(), smsMessage.getSignName(), smsMessage.getTemplateCode(), smsMessage.getTemplateParam().toString());
+            aBoolean = smsProcessor.send(smsMessage.getPhone(), smsMessage.getSignName(), smsMessage.getTemplateCode(), smsMessage.getTemplateParam().toString());
         } catch (Exception e) {
             log.error("CONSUMER SMS FAIL ==> {}", e.getMessage(), e);
-            return false;
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
         }
+        return aBoolean;
     }
 
     protected abstract String getRedisStreamKey();
+
+    protected abstract String getRedisStreamGroup();
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/AbstractSqlRedisReceiver.java

 package com.liquidnet.service.consumer.adam.receiver;
 
+import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.base.constant.MQConst;
 import com.liquidnet.service.consumer.adam.service.IBaseDao;
 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 java.util.HashMap;
+
 @Slf4j
 public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
     @Autowired
@@ -22,29 +27,39 @@ public abstract class AbstractSqlRedisReceiver implements StreamListener<String,
                 this.getRedisStreamKey(), message.getId(), message.getStream(), message.getValue());
 
         boolean result = this.consumerSqlDaoHandler(message.getValue().get("message"));
+        log.info("CONSUMER SMS RESULT:{} ==> MESSAGE_ID:{}", result, message.getId());
 
-        log.debug("CONSUMER SQL RESULT:{}", result);
-
-        // 消费成功确认，消息删除和消息确认是一个事务
-        if (result) {
-            log.info("CONSUMER SMS SUCC ==> MESSAGE_ID:{}", message.getId());
+        try {
+            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+            stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
+        } catch (Exception e) {
+            log.error("CONSUMER SQL RESULT:{} ==> DEL_REDIS_QUEUE_MSG_EXCEPTION[MESSAGE_ID:{},MSG:{}]", result, message.getId(), JsonUtils.toJson(message), e);
+        } finally {
             try {
-                stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
-            } catch (Exception e) {
-                log.error("CONSUMER SMS SUCC ==> DEL_REDIS_QUEUE_MSG_EXCEPTION[MESSAGE_ID:{}]", message.getId(), e);
+                stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+            } catch (Exception ignored) {
             }
         }
     }
 
     private boolean consumerSqlDaoHandler(String msg) {
+        Boolean aBoolean = false;
         try {
             SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
-            return baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
         } catch (Exception e) {
             log.error("CONSUMER SMS FAIL ==> {}", e.getMessage(), e);
-            return false;
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
         }
+        return aBoolean;
     }
 
     protected abstract String getRedisStreamKey();
+
+    protected abstract String getRedisStreamGroup();
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/ConsumerAdamSmsNoticeRdsReceiver.java

@@ -9,4 +9,9 @@ public class ConsumerAdamSmsNoticeRdsReceiver extends AbstractSmsRedisReceiver {
     protected String getRedisStreamKey() {
         return MQConst.AdamQueue.SMS_NOTICE.getKey();
     }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.AdamQueue.SMS_NOTICE.getGroup();
+    }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/ConsumerAdamSqlUCenterRdsReceiver.java

@@ -9,4 +9,9 @@ public class ConsumerAdamSqlUCenterRdsReceiver extends AbstractSqlRedisReceiver
     protected String getRedisStreamKey() {
         return MQConst.AdamQueue.SQL_UCENTER.getKey();
     }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.AdamQueue.SQL_UCENTER.getGroup();
+    }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/ConsumerAdamSqlUMemberRdsReceiver.java

@@ -9,4 +9,9 @@ public class ConsumerAdamSqlUMemberRdsReceiver extends AbstractSqlRedisReceiver
     protected String getRedisStreamKey() {
         return MQConst.AdamQueue.SQL_UMEMBER.getKey();
     }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.AdamQueue.SQL_UMEMBER.getGroup();
+    }
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/ConsumerAdamSqlURegisterRdsReceiver.java

@@ -9,4 +9,9 @@ public class ConsumerAdamSqlURegisterRdsReceiver extends AbstractSqlRedisReceive
     protected String getRedisStreamKey() {
         return MQConst.AdamQueue.SQL_UREGISTER.getKey();
     }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.AdamQueue.SQL_UREGISTER.getGroup();
+    }
 }