+短信消费;

liquidnet-bus-common/liquidnet-common-sms/src/main/java/com/liquidnet/common/sms/processor/SmsProcessor.java

@@ -51,7 +51,13 @@ public class SmsProcessor {
 
             SendSmsResponseBody smsResponseBody = smsResponse.getBody();
 
-            return null != smsResponseBody && smsResponseBody.getCode().equalsIgnoreCase("OK");
+            boolean isSucc = null != smsResponseBody && "OK".equalsIgnoreCase(smsResponseBody.getCode());
+
+            if (!isSucc) {
+                log.warn("###短信服务发送失败[{}]", JsonUtils.toJson(smsResponse));
+            }
+
+            return isSucc;
         } catch (Exception e) {
             log.error("API处理异常", e);
         }
@@ -81,7 +87,13 @@ public class SmsProcessor {
 
             SendBatchSmsResponseBody batchSmsResponseBody = batchSmsResponse.getBody();
 
-            return null != batchSmsResponseBody && batchSmsResponseBody.getCode().equalsIgnoreCase("OK");
+            boolean isSucc = null != batchSmsResponseBody && "OK".equalsIgnoreCase(batchSmsResponseBody.getCode());
+
+            if (!isSucc) {
+                log.warn("###短信服务发送失败[{}]", JsonUtils.toJson(batchSmsResponse));
+            }
+
+            return isSucc;
         } catch (Exception e) {
             log.error("API处理异常", e);
         }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/service/processor/ConsumerAdamSmsProcessor.java

+package com.liquidnet.service.consumer.adam.service.processor;
+
+import com.liquidnet.common.mq.constant.MQConst;
+import com.liquidnet.common.sms.processor.SmsProcessor;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SmsMessage;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.annotation.Exchange;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.QueueBinding;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+
+/**
+ * ConsumerAdamSmsProcessor.class
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2021/7/13
+ */
+@Slf4j
+@Component
+public class ConsumerAdamSmsProcessor {
+    @Resource
+    SmsProcessor smsProcessor;
+
+    private void consumerSmsSendHandler(Message msg, Channel channel) {
+        MessageProperties properties = msg.getMessageProperties();
+        String consumerQueue = properties.getConsumerQueue();
+        long deliveryTag = properties.getDeliveryTag();
+        log.info("CONSUMER SMS ==> [consumerQueue:{},deliveryTag:{}]", consumerQueue, deliveryTag);
+        String msgBody = new String(msg.getBody());
+        log.debug("CONSUMER SMS ==> Preparing:{}", msgBody);
+        try {
+            SmsMessage smsMessage = JsonUtils.fromJson(msgBody, SmsMessage.class);
+            boolean result = smsProcessor.aliyunDysmsSend(smsMessage.getPhone(), smsMessage.getSignName(), smsMessage.getTemplateCode(), smsMessage.getTemplateParam().toString());
+            log.debug("CONSUMER SMS result of execution:{}", result);
+            if (result) {
+                channel.basicAck(deliveryTag, false);
+            } else {
+                log.warn("###CONSUMER SMS[consumerQueue:{},deliveryTag={},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody);
+                channel.basicAck(deliveryTag, false);
+            }
+        } catch (IOException e) {
+            log.error("CONSUMER SMS[consumerQueue:{},deliveryTag:{},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody, e);
+        }
+    }
+
+    /* ================================================================== ｜ 短信验证码 */
+
+    @RabbitListener(
+            bindings = @QueueBinding(
+                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
+                    key = MQConst.RK_SMS_CODE,
+                    value = @Queue(MQConst.QUEUES_SMS_CODE)
+            ),
+            concurrency = "20"
+    )
+    public void consumerSqlForSmsCode(Message msg, Channel channel) {
+        this.consumerSmsSendHandler(msg, channel);
+    }
+
+    /* ================================================================== ｜ 短信通知 */
+
+    @RabbitListener(
+            bindings = @QueueBinding(
+                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
+                    key = MQConst.RK_SMS_NOTICE,
+                    value = @Queue(MQConst.QUEUES_SMS_NOTICE)
+            ),
+            concurrency = "10"
+    )
+    public void consumerSqlForSmsNotice(Message msg, Channel channel) {
+        this.consumerSmsSendHandler(msg, channel);
+    }
+
+
+    /* ================================================================== ｜  */
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/service/processor/ConsumerAdamUCenterProcessor.java

 package com.liquidnet.service.consumer.adam.service.processor;
 
 import com.liquidnet.common.mq.constant.MQConst;
-import com.liquidnet.common.sms.processor.SmsProcessor;
 import com.liquidnet.commons.lang.util.JsonUtils;
-import com.liquidnet.service.base.SmsMessage;
 import com.liquidnet.service.base.SqlMapping;
 import com.liquidnet.service.consumer.adam.service.IBaseDao;
 import com.rabbitmq.client.Channel;
@@ -30,8 +28,6 @@ import java.io.IOException;
 public class ConsumerAdamUCenterProcessor {
     @Resource
     IBaseDao baseDao;
-    @Resource
-    SmsProcessor smsProcessor;
 
 //    @RabbitListener(bindings = @QueueBinding(
 //            exchange = @Exchange(MQConst.EXCHANGES_LIQUIDNET_SQL), key = MQConst.ROUTING_KEY_SQL,
@@ -83,28 +79,6 @@ public class ConsumerAdamUCenterProcessor {
         }
     }
 
-    private void consumerSmsSendHandler(Message msg, Channel channel) {
-        MessageProperties properties = msg.getMessageProperties();
-        String consumerQueue = properties.getConsumerQueue();
-        long deliveryTag = properties.getDeliveryTag();
-        log.info("CONSUMER SMS ==> [consumerQueue:{},deliveryTag:{}]", consumerQueue, deliveryTag);
-        String msgBody = new String(msg.getBody());
-        log.debug("CONSUMER SMS ==> Preparing:{}", msgBody);
-        try {
-            SmsMessage smsMessage = JsonUtils.fromJson(msgBody, SmsMessage.class);
-            boolean result = smsProcessor.aliyunDysmsSend(smsMessage.getPhone(), smsMessage.getSignName(), smsMessage.getTemplateCode(), smsMessage.getTemplateParam().toString());
-            log.debug("CONSUMER SMS result of execution:{}", result);
-            if (result) {
-                channel.basicAck(deliveryTag, false);
-            } else {
-                log.warn("###CONSUMER SMS[consumerQueue:{},deliveryTag={},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody);
-                channel.basicAck(deliveryTag, false);
-            }
-        } catch (IOException e) {
-            log.error("CONSUMER SMS[consumerQueue:{},deliveryTag:{},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody, e);
-        }
-    }
-
     /* ================================================================== ｜ 用户注册 */
 
     @RabbitListener(
@@ -146,35 +120,4 @@ public class ConsumerAdamUCenterProcessor {
     public void consumerSqlForUMember(Message msg, Channel channel) {
         this.consumerSqlDaoHandler(msg, channel);
     }
-
-    /* ================================================================== ｜ 短信验证码 */
-
-    @RabbitListener(
-            bindings = @QueueBinding(
-                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
-                    key = MQConst.RK_SMS_CODE,
-                    value = @Queue(MQConst.QUEUES_SMS_CODE)
-            ),
-            concurrency = "10"
-    )
-    public void consumerSqlForSmsCode(Message msg, Channel channel) {
-        this.consumerSmsSendHandler(msg, channel);
-    }
-
-    /* ================================================================== ｜ 短信通知 */
-
-    @RabbitListener(
-            bindings = @QueueBinding(
-                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
-                    key = MQConst.RK_SMS_NOTICE,
-                    value = @Queue(MQConst.QUEUES_SMS_NOTICE)
-            ),
-            concurrency = "5"
-    )
-    public void consumerSqlForSmsNotice(Message msg, Channel channel) {
-        this.consumerSmsSendHandler(msg, channel);
-    }
-
-
-    /* ================================================================== ｜  */
 }

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/pom.xml

@@ -31,6 +31,10 @@
             <artifactId>liquidnet-common-cache-redisson</artifactId>
             <version>1.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.liquidnet</groupId>
             <artifactId>liquidnet-common-cache-redis</artifactId>
@@ -42,8 +46,9 @@
             <version>1.0-SNAPSHOT</version>
         </dependency>
         <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-common-sms</artifactId>
+            <version>1.0-SNAPSHOT</version>
         </dependency>
     </dependencies>
 

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/service/processor/ConsumerKylinSmsProcessor.java

+package com.liquidnet.service.consumer.kylin.service.processor;
+
+import com.liquidnet.common.mq.constant.MQConst;
+import com.liquidnet.common.sms.processor.SmsProcessor;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SmsMessage;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.annotation.Exchange;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.QueueBinding;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+
+/**
+ * ConsumerAdamSmsProcessor.class
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2021/7/13
+ */
+@Slf4j
+@Component
+public class ConsumerKylinSmsProcessor {
+    @Resource
+    SmsProcessor smsProcessor;
+
+    private void consumerSmsSendHandler(Message msg, Channel channel) {
+        MessageProperties properties = msg.getMessageProperties();
+        String consumerQueue = properties.getConsumerQueue();
+        long deliveryTag = properties.getDeliveryTag();
+        log.info("CONSUMER SMS ==> [consumerQueue:{},deliveryTag:{}]", consumerQueue, deliveryTag);
+        String msgBody = new String(msg.getBody());
+        log.debug("CONSUMER SMS ==> Preparing:{}", msgBody);
+        try {
+            SmsMessage smsMessage = JsonUtils.fromJson(msgBody, SmsMessage.class);
+            boolean result = smsProcessor.aliyunDysmsSend(smsMessage.getPhone(), smsMessage.getSignName(), smsMessage.getTemplateCode(), smsMessage.getTemplateParam().toString());
+            log.debug("CONSUMER SMS result of execution:{}", result);
+            if (result) {
+                channel.basicAck(deliveryTag, false);
+            } else {
+                log.warn("###CONSUMER SMS[consumerQueue:{},deliveryTag={},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody);
+                channel.basicAck(deliveryTag, false);
+            }
+        } catch (IOException e) {
+            log.error("CONSUMER SMS[consumerQueue:{},deliveryTag:{},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody, e);
+        }
+    }
+
+    /* ================================================================== ｜ 短信验证码 */
+
+    @RabbitListener(
+            bindings = @QueueBinding(
+                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
+                    key = MQConst.RK_SMS_CODE,
+                    value = @Queue(MQConst.QUEUES_SMS_CODE)
+            ),
+            concurrency = "20"
+    )
+    public void consumerSqlForSmsCode(Message msg, Channel channel) {
+        this.consumerSmsSendHandler(msg, channel);
+    }
+
+    /* ================================================================== ｜ 短信通知 */
+
+    @RabbitListener(
+            bindings = @QueueBinding(
+                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
+                    key = MQConst.RK_SMS_NOTICE,
+                    value = @Queue(MQConst.QUEUES_SMS_NOTICE)
+            ),
+            concurrency = "10"
+    )
+    public void consumerSqlForSmsNotice(Message msg, Channel channel) {
+        this.consumerSmsSendHandler(msg, channel);
+    }
+
+
+    /* ================================================================== ｜  */
+}