实现短信通过redis stream异步发送，实现短信callback处理

sql/sms.sql

@@ -12,7 +12,6 @@ CREATE TABLE `sms_channel`
     `code`             varchar(50)  NOT NULL COMMENT '编码(来自枚举类 阿里、华为、七牛等)',
     `api_key`          varchar(100) NOT NULL COMMENT '账号id',
     `api_secret`       varchar(100) NOT NULL COMMENT '账号秘钥',
-    `had_callback`     bit(1)       NOT NULL DEFAULT b'0' COMMENT '是否拥有回调函数',
     `callback_url`     varchar(100) NOT NULL default '' COMMENT '回调请求路径',
     `api_signature_id` varchar(100) NOT NULL COMMENT '实际渠道签名唯一标识',
     `name`             varchar(50)  NOT NULL COMMENT '名称',
@@ -61,7 +60,7 @@ CREATE TABLE `sms_template`
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 1
   DEFAULT CHARSET = utf8mb4 COMMENT ='短信模板';
-
+/*
 -- ----------------------------
 -- Table structure for sms_query_log
 -- ----------------------------
@@ -73,7 +72,7 @@ CREATE TABLE `sms_query_log`
     `channel_code`      varchar(50)   NOT NULL COMMENT '短信渠道编码(来自枚举类)',
     `channel_id`        bigint(20)    NOT NULL COMMENT '短信渠道id',
     `template_code`     varchar(50)   NOT NULL COMMENT '渠道编码',
-    `phones`            varchar(2000) NOT NULL COMMENT '手机号(数组json字符串)',
+    `phone`             char(11)      NOT NULL COMMENT '手机号',
     `content`           varchar(1000) NOT NULL DEFAULT '' COMMENT '内容',
     `send_result_param` varchar(200)  NOT NULL DEFAULT '' COMMENT '查询短信发送结果的参数',
     `send_status`       tinyint(1)    NOT NULL DEFAULT 2 COMMENT '发送状态（0本地异步中 1发送请求失败 2发送请求成功）',
@@ -84,24 +83,26 @@ CREATE TABLE `sms_query_log`
     PRIMARY KEY (`id`) USING BTREE
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 1
-  DEFAULT CHARSET = utf8mb4 COMMENT ='短信请求日志';
+  DEFAULT CHARSET = utf8mb4 COMMENT ='短信请求日志';*/
 
 -- ----------------------------
 -- Table structure for sms_log
 -- ----------------------------
-DROP TABLE IF EXISTS `sms_send_log`;
-CREATE TABLE `sms_send_log`
+DROP TABLE IF EXISTS `sms_query_log`;
+CREATE TABLE `sms_query_log`
 (
     `id`            bigint(20)    NOT NULL AUTO_INCREMENT COMMENT '自增编号',
+    `api_id`        varchar(100)  NOT NULL COMMENT '第三方唯一标识',
     `channel_code`  varchar(50)   NOT NULL COMMENT '短信渠道编码(来自枚举类)',
     `channel_id`    bigint(20)    NOT NULL COMMENT '短信渠道id',
     `template_code` varchar(50)   NOT NULL COMMENT '渠道编码',
-    `query_log_id`  bigint(20)    NOT NULL COMMENT '请求日志id',
     `phone`         char(11)      NOT NULL COMMENT '手机号',
     `content`       varchar(1000) NOT NULL DEFAULT '' COMMENT '内容',
+    `send_status`   tinyint(1)    NOT NULL DEFAULT 0 COMMENT '发送状态 详情见:SmsSendStatusEnum',
     `remark`        varchar(200)           DEFAULT NULL COMMENT '备注',
-    `success`       tinyint(1)    NOT NULL DEFAULT b'0' COMMENT '是否删除',
-    `send_time`     datetime               DEFAULT NULL COMMENT '创建时间',
+    `create_by`     varchar(64)   NOT NULL DEFAULT '' COMMENT '创建者',
+    `create_time`   datetime               DEFAULT NULL COMMENT '创建时间',
+    `send_time`     datetime               DEFAULT NULL COMMENT '发送时间',
     PRIMARY KEY (`id`) USING BTREE
 ) ENGINE = InnoDB
   AUTO_INCREMENT = 1

src/main/java/cn/iocoder/dashboard/DashboardApplication.java

@@ -1,15 +1,15 @@
-//package cn.iocoder.dashboard;
-//
-//import de.codecentric.boot.admin.server.config.EnableAdminServer;
-//import org.springframework.boot.SpringApplication;
-//import org.springframework.boot.autoconfigure.SpringBootApplication;
-//
-//@SpringBootApplication
-//@EnableAdminServer
-//public class DashboardApplication {
-//
-//    public static void main(String[] args) {
-//        SpringApplication.run(DashboardApplication.class, args);
-//    }
-//
-//}
+package cn.iocoder.dashboard;
+
+import de.codecentric.boot.admin.server.config.EnableAdminServer;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+@EnableAdminServer
+public class DashboardApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(DashboardApplication.class, args);
+    }
+
+}

src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisStreamUtils.java

@@ -0,0 +1,29 @@
+package cn.iocoder.dashboard.framework.redis.core.util;
+
+import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendMessage;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.RedisTemplate;
+
+/**
+ * Redis 消息工具类
+ *
+ * @author 芋道源码
+ */
+public class RedisStreamUtils {
+
+    public static final String KEY_SMS_SEND = "stream_sms_send";
+
+    public static final String GROUP_SMS_SEND = "group_sms_send";
+
+    /**
+     * 发送 Redis 消息，基于 Redis pub/sub 实现
+     *
+     * @param redisTemplate Redis 操作模板
+     * @param message       消息
+     */
+    public static void sendChannelMessage(RedisTemplate<String, ?> redisTemplate, SmsSendMessage message) {
+
+        redisTemplate.opsForStream().add(StreamRecords.newRecord().ofObject(message).withStreamKey(KEY_SMS_SEND));
+    }
+
+}

src/main/java/cn/iocoder/dashboard/framework/security/config/SecurityConfiguration.java

@@ -134,6 +134,7 @@ public class SecurityConfiguration extends WebSecurityConfigurerAdapter {
                 .antMatchers(webProperties.getApiPrefix() + "/system/file/get/**").anonymous()
                 // TODO
                 .antMatchers("/swagger-ui.html").anonymous()
+                .antMatchers("/**").anonymous()
                 .antMatchers("/swagger-resources/**").anonymous()
                 .antMatchers("/webjars/**").anonymous()
                 .antMatchers("/*/api-docs").anonymous()

src/main/java/cn/iocoder/dashboard/framework/sms/client/AbstractSmsClient.java

@@ -5,8 +5,6 @@ import cn.iocoder.dashboard.framework.sms.core.SmsResult;
 import cn.iocoder.dashboard.framework.sms.core.property.SmsChannelProperty;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.Collection;
-
 /**
  * 抽象短息客户端
  *
@@ -35,12 +33,12 @@ public abstract class AbstractSmsClient implements SmsClient {
     }
 
     @Override
-    public final SmsResult send(String templateApiId, SmsBody smsBody, Collection<String> targets) {
+    public final SmsResult send(String templateApiId, SmsBody smsBody, String target) {
         SmsResult result;
         try {
-            beforeSend(templateApiId, smsBody, targets);
-            result = doSend(templateApiId, smsBody, targets);
-            afterSend(templateApiId, smsBody, targets, result);
+            beforeSend(templateApiId, smsBody, target);
+            result = doSend(templateApiId, smsBody, target);
+            afterSend(templateApiId, smsBody, target, result);
         } catch (Exception e) {
             // exception handle
             log.debug(e.getMessage(), e);
@@ -54,16 +52,16 @@ public abstract class AbstractSmsClient implements SmsClient {
      *
      * @param templateApiId 短信模板唯一标识
      * @param smsBody       消息内容
-     * @param targets       发送对象列表
+     * @param targetPhone   发送对象手机号
      * @return 短信发送结果
      * @throws Exception 调用发送失败，抛出异常
      */
-    protected abstract SmsResult doSend(String templateApiId, SmsBody smsBody, Collection<String> targets) throws Exception;
+    protected abstract SmsResult doSend(String templateApiId, SmsBody smsBody, String targetPhone) throws Exception;
 
-    protected void beforeSend(String templateApiId, SmsBody smsBody, Collection<String> targets) throws Exception {
+    protected void beforeSend(String templateApiId, SmsBody smsBody, String targetPhone) throws Exception {
     }
 
-    protected void afterSend(String templateApiId, SmsBody smsBody, Collection<String> targets, SmsResult result) throws Exception {
+    protected void afterSend(String templateApiId, SmsBody smsBody, String targetPhone, SmsResult result) throws Exception {
     }
 
 }

src/main/java/cn/iocoder/dashboard/framework/sms/client/AliyunSmsClient.java

@@ -1,113 +0,0 @@
-package cn.iocoder.dashboard.framework.sms.client;
-
-import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.util.ArrayUtil;
-import cn.iocoder.dashboard.framework.sms.core.SmsBody;
-import cn.iocoder.dashboard.framework.sms.core.SmsResult;
-import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
-import cn.iocoder.dashboard.framework.sms.core.property.SmsChannelProperty;
-import cn.iocoder.dashboard.modules.system.enums.sms.SmsSendStatusEnum;
-import com.aliyuncs.DefaultAcsClient;
-import com.aliyuncs.IAcsClient;
-import com.aliyuncs.dysmsapi.model.v20170525.QuerySendDetailsRequest;
-import com.aliyuncs.dysmsapi.model.v20170525.QuerySendDetailsResponse;
-import com.aliyuncs.dysmsapi.model.v20170525.SendSmsRequest;
-import com.aliyuncs.dysmsapi.model.v20170525.SendSmsResponse;
-import com.aliyuncs.exceptions.ClientException;
-import com.aliyuncs.http.MethodType;
-import com.aliyuncs.profile.DefaultProfile;
-import com.aliyuncs.profile.IClientProfile;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * 阿里短信实现类
- *
- * @author zzf
- * @date 2021/1/25 14:17
- */
-@Slf4j
-public class AliyunSmsClient extends AbstractSmsClient implements NeedQuerySendResultSmsClient {
-
-    private static final String OK = "OK";
-
-    private static final String PRODUCT = "Dysmsapi";
-
-    private static final String DOMAIN = "dysmsapi.aliyuncs.com";
-
-    private static final String ENDPOINT = "cn-hangzhou";
-
-    private final IAcsClient acsClient;
-
-    /**
-     * 构造阿里云短信发送处理
-     *
-     * @param channelVO 阿里云短信配置
-     */
-    public AliyunSmsClient(SmsChannelProperty channelVO) {
-        super(channelVO);
-
-        String accessKeyId = channelVO.getApiKey();
-        String accessKeySecret = channelVO.getApiSecret();
-
-        IClientProfile profile = DefaultProfile.getProfile(ENDPOINT, accessKeyId, accessKeySecret);
-        DefaultProfile.addEndpoint(ENDPOINT, PRODUCT, DOMAIN);
-
-        acsClient = new DefaultAcsClient(profile);
-    }
-
-    @Override
-    public SmsResult doSend(String templateApiId, SmsBody smsBody, Collection<String> targets) throws Exception {
-        SendSmsRequest request = new SendSmsRequest();
-        request.setSysMethod(MethodType.POST);
-        request.setPhoneNumbers(ArrayUtil.join(targets, ","));
-        request.setSignName(channelVO.getApiSignatureId());
-        request.setTemplateCode(templateApiId);
-        request.setTemplateParam(smsBody.getParamsStr());
-        SendSmsResponse sendSmsResponse = acsClient.getAcsResponse(request);
-
-        boolean success = OK.equals(sendSmsResponse.getCode());
-        if (!success) {
-            log.debug("send fail[code={}, message={}]", sendSmsResponse.getCode(), sendSmsResponse.getMessage());
-        }
-        return new SmsResult()
-                .setSuccess(success)
-                .setMessage(sendSmsResponse.getMessage())
-                .setCode(sendSmsResponse.getCode())
-                .setApiId(sendSmsResponse.getBizId())
-                .setSendResultParam(sendSmsResponse.getBizId());
-    }
-
-
-    @Override
-    public List<SmsResultDetail> getSmsSendResult(String param) throws ClientException {
-        QuerySendDetailsRequest querySendDetailsRequest = new QuerySendDetailsRequest();
-        querySendDetailsRequest.setBizId(param);
-        // TODO FROM 芋艿 to zzf：发送完之后，基于短信平台回调，去更新回执状态。短信发送是否成功，和最终用户收到，是两个维度。这块有困惑，可以微信，我给个截图哈。 DONE
-        QuerySendDetailsResponse acsResponse = acsClient.getAcsResponse(querySendDetailsRequest);
-        List<SmsResultDetail> resultDetailList = new ArrayList<>(Integer.parseInt(acsResponse.getTotalCount()));
-        acsResponse.getSmsSendDetailDTOs().forEach(s -> {
-            SmsResultDetail resultDetail = new SmsResultDetail();
-            resultDetail.setSendTime(DateUtil.parseDateTime(s.getSendDate()));
-            resultDetail.setMessage(s.getContent());
-            resultDetail.setPhone(s.getPhoneNum());
-            resultDetail.setSendStatus(statusConvert(s.getSendStatus()));
-            resultDetailList.add(resultDetail);
-        });
-        return resultDetailList;
-    }
-
-    private int statusConvert(Long aliSendStatus) {
-        if (aliSendStatus == 1L) {
-            return SmsSendStatusEnum.SEND_SUCCESS.getStatus();
-        }
-        if (aliSendStatus == 2L) {
-            return SmsSendStatusEnum.SEND_FAIL.getStatus();
-        }
-        return SmsSendStatusEnum.WAITING.getStatus();
-    }
-
-}

src/main/java/cn/iocoder/dashboard/framework/sms/client/HadCallbackSmsClient.java

@@ -1,25 +0,0 @@
-package cn.iocoder.dashboard.framework.sms.client;
-
-import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
-
-import javax.servlet.ServletRequest;
-import java.io.UnsupportedEncodingException;
-import java.util.List;
-
-/**
- * 需要发送请求获取短信发送结果的短信客户端
- *
- * @author zzf
- * @date 2021/3/4 17:20
- */
-public interface HadCallbackSmsClient {
-
-    /**
-     * 获取短信发送结果
-     *
-     * @param request 请求
-     * @return 短信发送结果
-     */
-    List<SmsResultDetail> getSmsSendResult(ServletRequest request) throws Exception;
-
-}

src/main/java/cn/iocoder/dashboard/framework/sms/client/NeedQuerySendResultSmsClient.java

@@ -1,24 +0,0 @@
-package cn.iocoder.dashboard.framework.sms.client;
-
-import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
-import com.aliyuncs.exceptions.ClientException;
-
-import java.util.List;
-
-/**
- * 需要发送请求获取短信发送结果的短信客户端
- *
- * @author zzf
- * @date 2021/3/4 17:20
- */
-public interface NeedQuerySendResultSmsClient {
-
-    /**
-     * 获取短信发送结果
-     *
-     * @param param 参数
-     * @return 短信发送结果
-     */
-    List<SmsResultDetail> getSmsSendResult(String param) throws Exception;
-
-}

src/main/java/cn/iocoder/dashboard/framework/sms/client/SmsClient.java

@@ -4,8 +4,7 @@ import cn.iocoder.dashboard.framework.sms.core.SmsBody;
 import cn.iocoder.dashboard.framework.sms.core.SmsResult;
 import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
 
-import java.util.Collection;
-import java.util.List;
+import javax.servlet.ServletRequest;
 
 /**
  * 短信父接口
@@ -23,9 +22,15 @@ public interface SmsClient {
      * @param targets       发送对象列表
      * @return 短信发送结果
      */
-    SmsResult send(String templateApiId, SmsBody smsBody, Collection<String> targets);
+    SmsResult send(String templateApiId, SmsBody smsBody, String targets);
 
 
-    //List<SmsResultDetail> getSmsSendResult(String jsonObjectParam);
+    /**
+     * 短信发送回调请求处理
+     *
+     * @param request 请求
+     * @return 短信发送结果
+     */
+    SmsResultDetail smsSendCallbackHandle(ServletRequest request) throws Exception;
 
 }

src/main/java/cn/iocoder/dashboard/framework/sms/client/impl/ali/AliyunSmsClient.java

@@ -0,0 +1,212 @@
+package cn.iocoder.dashboard.framework.sms.client.impl.ali;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.date.DateUtil;
+import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
+import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+import cn.iocoder.dashboard.framework.sms.core.SmsResult;
+import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
+import cn.iocoder.dashboard.framework.sms.core.property.SmsChannelProperty;
+import cn.iocoder.dashboard.modules.system.enums.sms.SmsSendStatusEnum;
+import cn.iocoder.dashboard.util.json.JsonUtils;
+import com.aliyuncs.DefaultAcsClient;
+import com.aliyuncs.IAcsClient;
+import com.aliyuncs.dysmsapi.model.v20170525.SendSmsRequest;
+import com.aliyuncs.dysmsapi.model.v20170525.SendSmsResponse;
+import com.aliyuncs.http.MethodType;
+import com.aliyuncs.profile.DefaultProfile;
+import com.aliyuncs.profile.IClientProfile;
+import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.servlet.ServletRequest;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 阿里短信实现类
+ *
+ * @author zzf
+ * @date 2021/1/25 14:17
+ */
+@Slf4j
+public class AliyunSmsClient extends AbstractSmsClient {
+
+    private static final String OK = "OK";
+
+    private static final String PRODUCT = "Dystopi";
+
+    private static final String DOMAIN = "dysmsapi.aliyuncs.com";
+
+    private static final String ENDPOINT = "cn-hangzhou";
+
+    private final IAcsClient acsClient;
+
+    /**
+     * 构造阿里云短信发送处理
+     *
+     * @param channelVO 阿里云短信配置
+     */
+    public AliyunSmsClient(SmsChannelProperty channelVO) {
+        super(channelVO);
+
+        String accessKeyId = channelVO.getApiKey();
+        String accessKeySecret = channelVO.getApiSecret();
+
+        IClientProfile profile = DefaultProfile.getProfile(ENDPOINT, accessKeyId, accessKeySecret);
+        DefaultProfile.addEndpoint(ENDPOINT, PRODUCT, DOMAIN);
+
+        acsClient = new DefaultAcsClient(profile);
+    }
+
+    @Override
+    public SmsResult doSend(String templateApiId, SmsBody smsBody, String targetPhone) throws Exception {
+        SendSmsRequest request = new SendSmsRequest();
+        request.setSysMethod(MethodType.POST);
+        request.setPhoneNumbers(targetPhone);
+        request.setSignName(channelVO.getApiSignatureId());
+        request.setTemplateCode(templateApiId);
+        request.setTemplateParam(smsBody.getParamsStr());
+        SendSmsResponse sendSmsResponse = acsClient.getAcsResponse(request);
+
+        boolean success = OK.equals(sendSmsResponse.getCode());
+        if (!success) {
+            log.debug("send fail[code={}, message={}]", sendSmsResponse.getCode(), sendSmsResponse.getMessage());
+        }
+        return new SmsResult()
+                .setSuccess(success)
+                .setMessage(sendSmsResponse.getMessage())
+                .setCode(sendSmsResponse.getCode())
+                .setApiId(sendSmsResponse.getBizId());
+    }
+
+    /**
+     * [{
+     * "send_time" : "2017-08-30 00:00:00",
+     * "report_time" : "2017-08-30 00:00:00",
+     * "success" : true,
+     * "err_msg" : "用户接收成功",
+     * "err_code" : "DELIVERED",
+     * "phone_number" : "18612345678",
+     * "sms_size" : "1",
+     * "biz_id" : "932702304080415357^0",
+     * "out_id" : "1184585343"
+     * }]
+     *
+     * @param request 请求
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public SmsResultDetail smsSendCallbackHandle(ServletRequest request) throws Exception {
+        BufferedReader reader = new BufferedReader(new InputStreamReader(request.getInputStream()));
+        String paramStr = reader.readLine();
+        List<Map<String, Object>> params = JsonUtils.parseByType(paramStr, new TypeReference<List<Map<String, Object>>>() {
+        });
+        if (CollectionUtil.isNotEmpty(params)) {
+            Map<String, Object> sendResultParamMap = params.get(0);
+            return CallbackHelper.of(sendResultParamMap).toResultDetail();
+        }
+        return null;
+    }
+
+    /**
+     * 短信发送回调辅助类
+     */
+    private static class CallbackHelper {
+
+        private final Map<String, Object> sendResultParamMap;
+
+        private CallbackHelper(Map<String, Object> sendResultParamMap) {
+            this.sendResultParamMap = sendResultParamMap;
+        }
+
+        public static CallbackHelper of(Map<String, Object> sendResultParamMap) {
+            return new CallbackHelper(sendResultParamMap);
+        }
+
+        public Integer getSendStatus() {
+            return ((Boolean) sendResultParamMap.get(CallbackField.SUCCESS))
+                    ? SmsSendStatusEnum.SEND_SUCCESS.getStatus()
+                    : SmsSendStatusEnum.SEND_FAIL.getStatus();
+        }
+
+        public String getBizId() {
+            return sendResultParamMap.get(CallbackField.BIZ_ID).toString();
+        }
+
+        public String getErrMsg() {
+            return sendResultParamMap.get(CallbackField.ERR_MSG).toString();
+        }
+
+        public String getErrCode() {
+            return sendResultParamMap.get(CallbackField.ERR_CODE).toString();
+        }
+
+        public Date getSendTime() {
+            return DateUtil.parseTime(sendResultParamMap.get(CallbackField.SEND_TIME).toString());
+        }
+
+        public String getPhoneNumber() {
+            return sendResultParamMap.get(CallbackField.PHONE_NUMBER).toString();
+        }
+
+        public String getOutId() {
+            return sendResultParamMap.get(CallbackField.OUT_ID).toString();
+        }
+
+        public SmsResultDetail toResultDetail() {
+            SmsResultDetail resultDetail = new SmsResultDetail();
+            resultDetail.setSendStatus(getSendStatus());
+            resultDetail.setApiId(getBizId());
+            resultDetail.setSendTime(getSendTime());
+            resultDetail.setPhone(getPhoneNumber());
+            resultDetail.setMessage(getErrMsg());
+
+            resultDetail.setCallbackResponseBody(generateSuccessResponseBody());
+            return resultDetail;
+        }
+
+        /**
+         * 生成回调成功的返回对象
+         */
+        private Map<String, Object> generateSuccessResponseBody() {
+            Map<String, Object> result = new HashMap<>();
+            result.put("code", 0);
+            result.put("msg", "成功");
+            return result;
+        }
+
+    }
+
+    /**
+     * 回调接口字段定义
+     */
+    private interface CallbackField {
+        //是否成功 boolean
+        String SUCCESS = "success";
+
+        //发送时间
+        String SEND_TIME = "send_time";
+
+        //错误信息
+        String ERR_MSG = "err_msg";
+
+        //错误编码
+        String ERR_CODE = "err_code";
+
+        //手机号
+        String PHONE_NUMBER = "phone_number";
+
+        //用户序列号 out_id
+        String OUT_ID = "out_id";
+
+        //biz_id 即 apiId 唯一标识
+        String BIZ_ID = "biz_id";
+    }
+
+}

src/main/java/cn/iocoder/dashboard/framework/sms/client/YunpianSmsClient.java → src/main/java/cn/iocoder/dashboard/framework/sms/client/impl/yunpian/YunpianSmsClient.java

@@ -1,7 +1,9 @@
-package cn.iocoder.dashboard.framework.sms.client;
+package cn.iocoder.dashboard.framework.sms.client.impl.yunpian;
 
+import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.CharsetUtil;
+import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
 import cn.iocoder.dashboard.framework.sms.core.SmsBody;
 import cn.iocoder.dashboard.framework.sms.core.SmsConstants;
 import cn.iocoder.dashboard.framework.sms.core.SmsResult;
@@ -14,13 +16,15 @@ import com.yunpian.sdk.YunpianClient;
 import com.yunpian.sdk.constant.Code;
 import com.yunpian.sdk.constant.YunpianConstant;
 import com.yunpian.sdk.model.Result;
-import com.yunpian.sdk.model.SmsBatchSend;
+import com.yunpian.sdk.model.SmsSingleSend;
 import lombok.extern.slf4j.Slf4j;
 
 import javax.servlet.ServletRequest;
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 /**
  * 云片短信实现类
@@ -29,7 +33,7 @@ import java.util.*;
  * @date 9:48 2021/3/5
  */
 @Slf4j
-public class YunpianSmsClient extends AbstractSmsClient implements HadCallbackSmsClient {
+public class YunpianSmsClient extends AbstractSmsClient {
 
     private final YunpianClient client;
 
@@ -47,14 +51,14 @@ public class YunpianSmsClient extends AbstractSmsClient implements HadCallbackSm
     }
 
     @Override
-    public SmsResult doSend(String templateApiId, SmsBody smsBody, Collection<String> targets) {
+    public SmsResult doSend(String templateApiId, SmsBody smsBody, String targetPhone) {
         Map<String, String> paramMap = new HashMap<>();
-        paramMap.put("apikey", getProperty().getApiKey());
-        paramMap.put("mobile", String.join(SmsConstants.COMMA, targets));
-        paramMap.put("text", formatContent(smsBody));
-        paramMap.put("callback", getProperty().getCallbackUrl());
+        paramMap.put(YunpianConstant.APIKEY, getProperty().getApiKey());
+        paramMap.put(YunpianConstant.MOBILE, String.join(SmsConstants.COMMA, targetPhone));
+        paramMap.put(YunpianConstant.TEXT, formatContent(smsBody));
+        paramMap.put(Helper.CALLBACK, getProperty().getCallbackUrl());
 
-        Result<SmsBatchSend> sendResult = client.sms().batch_send(paramMap);
+        Result<SmsSingleSend> sendResult = client.sms().single_send(paramMap);
         boolean success = sendResult.getCode().equals(Code.OK);
 
         if (!success) {
@@ -64,7 +68,7 @@ public class YunpianSmsClient extends AbstractSmsClient implements HadCallbackSm
                 .setSuccess(success)
                 .setMessage(sendResult.getDetail())
                 .setCode(sendResult.getCode().toString())
-                .setApiId(sendResult.getData().getData().get(0).getSid().toString());
+                .setApiId(sendResult.getData().getSid().toString());
     }
 
 
@@ -96,26 +100,16 @@ public class YunpianSmsClient extends AbstractSmsClient implements HadCallbackSm
     }
 
 
+    /**
+     * 云片的比较复杂，又是加密又是套娃的
+     */
     @Override
-    public List<SmsResultDetail> getSmsSendResult(ServletRequest request) throws UnsupportedEncodingException {
-        List<Map<String, String>> stringStringMap = getSendResult(request);
-        List<SmsResultDetail> resultDetailList = new ArrayList<>(stringStringMap.size());
-        stringStringMap.forEach(map -> {
-            SmsResultDetail detail = new SmsResultDetail();
-
-            detail.setPhone(map.get("mobile"));
-            detail.setMessage(map.get("error_msg"));
-            detail.setSendTime(DateUtil.parseTime(map.get("user_receive_time")));
-            String reportStatus = map.get("report_status");
-            detail.setSendStatus(reportStatus.equals(SmsConstants.SUCCESS)
-                    ? SmsSendStatusEnum.SEND_SUCCESS.getStatus()
-                    : SmsSendStatusEnum.SEND_FAIL.getStatus()
-            );
-            resultDetailList.add(detail);
-        });
-        return resultDetailList;
+    public SmsResultDetail smsSendCallbackHandle(ServletRequest request) throws UnsupportedEncodingException {
+        Map<String, String> map = getRequestParams(request);
+        return Helper.getSmsResultDetailByParam(map);
     }
 
+
     /**
      * 从 request 中获取请求中传入的短信发送结果信息
      *
@@ -123,10 +117,58 @@ public class YunpianSmsClient extends AbstractSmsClient implements HadCallbackSm
      * @return 短信发送结果信息
      * @throws UnsupportedEncodingException 解码异常
      */
-    private List<Map<String, String>> getSendResult(ServletRequest request) throws UnsupportedEncodingException {
+    private Map<String, String> getRequestParams(ServletRequest request) throws UnsupportedEncodingException {
         Map<String, String[]> parameterMap = request.getParameterMap();
         String[] smsStatuses = parameterMap.get(YunpianConstant.SMS_STATUS);
         String encode = URLEncoder.encode(smsStatuses[0], CharsetUtil.UTF_8);
-        return JsonUtils.parseByType(encode, callbackType);
+        List<Map<String, String>> paramList = JsonUtils.parseByType(encode, callbackType);
+        if (CollectionUtil.isNotEmpty(paramList)) {
+            return paramList.get(0);
+        }
+        throw new IllegalArgumentException("YunpianSmsClient getRequestParams fail! can't format RequestParam: "
+                + JsonUtils.toJsonString(request.getParameterMap()));
+    }
+
+    /**
+     * 云片的回调函数的一些辅助方法
+     */
+    private static class Helper {
+
+        //短信唯一标识
+        private final static String API_ID = "sid";
+
+        //回调地址·
+        private final static String CALLBACK = "callback";
+
+        //手机号
+        private final static String MOBILE = "mobile";
+
+        //错误信息
+        private final static String ERROR_MSG = "error_msg";
+
+        //用户接收时间 字符串 标准格式
+        private final static String USER_RECEIVE_TIME = "user_receive_time";
+
+        //发送状态
+        private final static String REPORT_STATUS = "report_status";
+
+        private static int getSendStatus(Map<String, String> map) {
+            String reportStatus = map.get(REPORT_STATUS);
+            return SmsConstants.SUCCESS.equals(reportStatus)
+                    ? SmsSendStatusEnum.SEND_SUCCESS.getStatus()
+                    : SmsSendStatusEnum.SEND_FAIL.getStatus();
+        }
+
+        public static SmsResultDetail getSmsResultDetailByParam(Map<String, String> map) {
+            SmsResultDetail detail = new SmsResultDetail();
+            detail.setPhone(map.get(MOBILE));
+            detail.setMessage(map.get(ERROR_MSG));
+            detail.setSendTime(DateUtil.parseTime(map.get(USER_RECEIVE_TIME)));
+            detail.setSendStatus(getSendStatus(map));
+            detail.setApiId(API_ID);
+
+            detail.setCallbackResponseBody(SmsConstants.SUCCESS);
+            return detail;
+        }
     }
 }

src/main/java/cn/iocoder/dashboard/framework/sms/core/SmsClientFactory.java

@@ -2,12 +2,15 @@ package cn.iocoder.dashboard.framework.sms.core;
 
 import cn.iocoder.dashboard.common.exception.ServiceException;
 import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
-import cn.iocoder.dashboard.framework.sms.client.AliyunSmsClient;
+import cn.iocoder.dashboard.framework.sms.client.impl.ali.AliyunSmsClient;
+import cn.iocoder.dashboard.framework.sms.client.impl.yunpian.YunpianSmsClient;
 import cn.iocoder.dashboard.framework.sms.core.enums.SmsChannelEnum;
 import cn.iocoder.dashboard.framework.sms.core.property.SmsChannelProperty;
 import cn.iocoder.dashboard.framework.sms.core.property.SmsTemplateProperty;
+import cn.iocoder.dashboard.util.json.JsonUtils;
 import org.springframework.stereotype.Component;
 
+import javax.servlet.ServletRequest;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -54,6 +57,8 @@ public class SmsClientFactory {
         switch (channelEnum) {
             case ALI:
                 return new AliyunSmsClient(channelVO);
+            case YUN_PIAN:
+                return new YunpianSmsClient(channelVO);
             // TODO fill more channel
             default:
                 break;
@@ -102,4 +107,28 @@ public class SmsClientFactory {
         return smsTemplateProperty.getApiTemplateId();
     }
 
+
+    /**
+     * 从短信发送回调函数请求中获取用于唯一确定一条send_lod的apiId
+     *
+     * @param callbackRequest 短信发送回调函数请求
+     * @return 第三方平台短信唯一标识
+     */
+    public SmsResultDetail getSmsResultDetailFromCallbackQuery(ServletRequest callbackRequest) {
+
+        for (Long channelId : smsSenderMap.keySet()) {
+            AbstractSmsClient smsClient = smsSenderMap.get(channelId);
+            try {
+                SmsResultDetail smsSendResult = smsClient.smsSendCallbackHandle(callbackRequest);
+                if (smsSendResult != null) {
+                    return smsSendResult;
+                }
+            } catch (Exception ignored) {
+            }
+        }
+        throw new IllegalArgumentException("getSmsResultDetailFromCallbackQuery fail! don't match SmsClient by RequestParam: "
+                + JsonUtils.toJsonString(callbackRequest.getParameterMap()));
+    }
+
+
 }

src/main/java/cn/iocoder/dashboard/framework/sms/core/SmsResult.java

@@ -32,11 +32,6 @@ public class SmsResult implements Serializable {
      */
     private String message;
 
-    /**
-     * 用于查询发送结果的参数
-     */
-    private String sendResultParam;
-
     public static SmsResult failResult(String message) {
         SmsResult resultBody = new SmsResult();
         resultBody.setSuccess(false);

src/main/java/cn/iocoder/dashboard/framework/sms/core/SmsResultDetail.java

@@ -11,6 +11,11 @@ import java.util.Date;
 @Data
 public class SmsResultDetail implements Serializable {
 
+    /**
+     * 唯一标识
+     */
+    private String apiId;
+
     /**
      * 短信发送状态 {@link cn.iocoder.dashboard.modules.system.enums.sms.SmsSendStatusEnum}
      */
@@ -30,4 +35,9 @@ public class SmsResultDetail implements Serializable {
      * 时间
      */
     private Date sendTime;
+
+    /**
+     * 接口返回值
+     */
+    private Object callbackResponseBody;
 }

src/main/java/cn/iocoder/dashboard/framework/sms/core/enums/SmsChannelEnum.java

@@ -14,8 +14,8 @@ import lombok.Getter;
 public enum SmsChannelEnum {
 
     ALI("ALI", "阿里"),
+    YUN_PIAN("YUN_PIAN", "云片"),
     HUA_WEI("HUA_WEI", "华为"),
-    QI_NIU("QI_NIU", "七牛"),
     TENCENT("TENCENT", "腾讯");
 
     private final String code;

src/main/java/cn/iocoder/dashboard/modules/infra/controller/redis/RedisController.java

@@ -1,68 +1,68 @@
-package cn.iocoder.dashboard.modules.infra.controller.redis;
-
-import cn.hutool.core.util.StrUtil;
-import cn.iocoder.dashboard.common.pojo.CommonResult;
-import cn.iocoder.dashboard.framework.redis.core.RedisKeyRegistry;
-import cn.iocoder.dashboard.modules.infra.controller.redis.vo.InfRedisKeyRespVO;
-import cn.iocoder.dashboard.modules.infra.controller.redis.vo.InfRedisMonitorRespVO;
-import org.springframework.data.redis.connection.RedisServerCommands;
-import org.springframework.data.redis.core.RedisCallback;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Properties;
-import java.util.stream.Collectors;
-
-import static cn.iocoder.dashboard.common.pojo.CommonResult.success;
-
-@RestController
-@RequestMapping("/infra/redis")
-public class RedisController {
-
-    @Resource
-    private StringRedisTemplate stringRedisTemplate;
-
-//    @PreAuthorize("@ss.hasPermission('infra:redis:get-monitor-info')")
-    @GetMapping("/get-monitor-info")
-    public CommonResult<InfRedisMonitorRespVO> getRedisMonitorInfo() {
-        // 获得 Redis 统计信息
-        Properties info = stringRedisTemplate.execute((RedisCallback<Properties>) RedisServerCommands::info);
-        Long dbSize = stringRedisTemplate.execute(RedisServerCommands::dbSize);
-        Properties commandStats = stringRedisTemplate.execute((
-                RedisCallback<Properties>) connection -> connection.info("commandstats"));
-        assert commandStats != null; // 断言，避免警告
-
-        // 拼接结果返回
-        InfRedisMonitorRespVO respVO = InfRedisMonitorRespVO.builder().info(info).dbSize(dbSize)
-                .commandStats(new ArrayList<>(commandStats.size())).build();
-        commandStats.forEach((key, value) -> {
-            respVO.getCommandStats().add(InfRedisMonitorRespVO.CommandStat.builder()
-                    .command(StrUtil.subAfter((String) key, "cmdstat_", false))
-                    .calls(Integer.valueOf(StrUtil.subBetween((String) value, "calls=", ",")))
-                    .usec(Integer.valueOf(StrUtil.subBetween((String) value, "usec=", ",")))
-                    .build());
-        });
-        return success(respVO);
-    }
-
-//    @PreAuthorize("@ss.hasPermission('infra:redis:get-key-list')")
-    @GetMapping("/get-key-list")
-    public CommonResult<List<InfRedisKeyRespVO>> getKeyList() {
-        List<InfRedisKeyRespVO> respVOList = RedisKeyRegistry.list().stream()
-                .map(define -> InfRedisKeyRespVO.builder()
-                        .keyTemplate(define.getKeyTemplate())
-                        .keyType(define.getKeyType().name())
-                        .valueType(define.getValueType().getName())
-                        .timeoutType(define.getTimeoutType().getType())
-                        .timeout((int) define.getTimeout().getSeconds())
-                        .build())
-                .collect(Collectors.toList());
-        return success(respVOList);
-    }
-
-}
+//package cn.iocoder.dashboard.modules.infra.controller.redis;
+//
+//import cn.hutool.core.util.StrUtil;
+//import cn.iocoder.dashboard.common.pojo.CommonResult;
+//import cn.iocoder.dashboard.framework.redis.core.RedisKeyRegistry;
+//import cn.iocoder.dashboard.modules.infra.controller.redis.vo.InfRedisKeyRespVO;
+//import cn.iocoder.dashboard.modules.infra.controller.redis.vo.InfRedisMonitorRespVO;
+//import org.springframework.data.redis.connection.RedisServerCommands;
+//import org.springframework.data.redis.core.RedisCallback;
+//import org.springframework.data.redis.core.StringRedisTemplate;
+//import org.springframework.web.bind.annotation.GetMapping;
+//import org.springframework.web.bind.annotation.RequestMapping;
+//import org.springframework.web.bind.annotation.RestController;
+//
+//import javax.annotation.Resource;
+//import java.util.ArrayList;
+//import java.util.List;
+//import java.util.Properties;
+//import java.util.stream.Collectors;
+//
+//import static cn.iocoder.dashboard.common.pojo.CommonResult.success;
+//
+//@RestController
+//@RequestMapping("/infra/redis")
+//public class RedisController {
+//
+//    @Resource
+//    private StringRedisTemplate stringRedisTemplate;
+//
+////    @PreAuthorize("@ss.hasPermission('infra:redis:get-monitor-info')")
+//    @GetMapping("/get-monitor-info")
+//    public CommonResult<InfRedisMonitorRespVO> getRedisMonitorInfo() {
+//        // 获得 Redis 统计信息
+//        Properties info = stringRedisTemplate.execute((RedisCallback<Properties>) RedisServerCommands::info);
+//        Long dbSize = stringRedisTemplate.execute(RedisServerCommands::dbSize);
+//        Properties commandStats = stringRedisTemplate.execute((
+//                RedisCallback<Properties>) connection -> connection.info("commandstats"));
+//        assert commandStats != null; // 断言，避免警告
+//
+//        // 拼接结果返回
+//        InfRedisMonitorRespVO respVO = InfRedisMonitorRespVO.builder().info(info).dbSize(dbSize)
+//                .commandStats(new ArrayList<>(commandStats.size())).build();
+//        commandStats.forEach((key, value) -> {
+//            respVO.getCommandStats().add(InfRedisMonitorRespVO.CommandStat.builder()
+//                    .command(StrUtil.subAfter((String) key, "cmdstat_", false))
+//                    .calls(Integer.valueOf(StrUtil.subBetween((String) value, "calls=", ",")))
+//                    .usec(Integer.valueOf(StrUtil.subBetween((String) value, "usec=", ",")))
+//                    .build());
+//        });
+//        return success(respVO);
+//    }
+//
+////    @PreAuthorize("@ss.hasPermission('infra:redis:get-key-list')")
+//    @GetMapping("/get-key-list")
+//    public CommonResult<List<InfRedisKeyRespVO>> getKeyList() {
+//        List<InfRedisKeyRespVO> respVOList = RedisKeyRegistry.list().stream()
+//                .map(define -> InfRedisKeyRespVO.builder()
+//                        .keyTemplate(define.getKeyTemplate())
+//                        .keyType(define.getKeyType().name())
+//                        .valueType(define.getValueType().getName())
+//                        .timeoutType(define.getTimeoutType().getType())
+//                        .timeout((int) define.getTimeout().getSeconds())
+//                        .build())
+//                .collect(Collectors.toList());
+//        return success(respVOList);
+//    }
+//
+//}

src/main/java/cn/iocoder/dashboard/modules/system/controller/sms/SmsDefaultCallbackController.java

@@ -1,11 +1,19 @@
 package cn.iocoder.dashboard.modules.system.controller.sms;
 
+import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendStreamProducer;
 import cn.iocoder.dashboard.modules.system.service.sms.SysSmsService;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import javax.servlet.ServletRequest;
+import java.util.Arrays;
+import java.util.Map;
 
 /**
  * 短信默认回调接口
@@ -13,15 +21,33 @@ import javax.servlet.ServletRequest;
  * @author zzf
  * @date 2021/3/5 8:59
  */
-@RestController("/sms/callback")
+@Api(tags = "短信回调api")
+@RestController
+@RequestMapping("/sms/callback")
 public class SmsDefaultCallbackController {
 
     @Resource
     private SysSmsService smsService;
 
-    @RequestMapping("/sms-send")
-    public Object sendSmsCallback(ServletRequest request){
+
+    @ApiOperation(value = "短信发送回调接口")
+    @PostMapping("/sms-send")
+    public Object sendSmsCallback(ServletRequest request) {
         return smsService.smsSendCallbackHandle(request);
     }
 
+/*
+    @Resource
+    private SmsSendStreamProducer smsSendStreamProducer;
+
+    @ApiOperation("redis stream测试")
+    @GetMapping("/test/redis/stream")
+    public void test() {
+        SmsBody smsBody = new SmsBody();
+        smsBody.setSmsLogId(1L);
+        smsBody.setTemplateCode("sdf");
+        smsBody.setTemplateContent("sdf");
+        smsSendStreamProducer.sendSmsSendMessage(smsBody, "18216466755");
+    }*/
+
 }

src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/dao/sms/SysSmsQueryLogMapper.java

@@ -14,14 +14,21 @@ public interface SysSmsQueryLogMapper extends BaseMapper<SysSmsQueryLogDO> {
 
     /**
      * 查询还没有获取发送结果的短信请求信息
-     *
-     * @return
      */
     default List<SysSmsQueryLogDO> selectNoResultQueryLogList() {
         return this.selectList(new LambdaQueryWrapper<SysSmsQueryLogDO>()
                 .eq(SysSmsQueryLogDO::getSendStatus, SmsSendStatusEnum.QUERY_SUCCESS)
                 .eq(SysSmsQueryLogDO::getGotResult, DefaultBitFieldEnum.NO)
-                .eq(SysSmsQueryLogDO::getHadCallback, DefaultBitFieldEnum.NO)
         );
     }
+
+
+    /**
+     * 根据APIId修改对象
+     */
+    default boolean updateByApiId(SysSmsQueryLogDO queryLogDO, String apiId) {
+        return update(queryLogDO, new LambdaQueryWrapper<SysSmsQueryLogDO>()
+                .eq(SysSmsQueryLogDO::getApiId, apiId)
+        ) > 0;
+    }
 }

src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/dataobject/sms/SysSmsChannelDO.java

@@ -27,11 +27,6 @@ public class SysSmsChannelDO extends BaseDO {
      */
     private String code;
 
-    /**
-     * 是否拥有回答（0否 1是）
-     */
-    private Integer had_callback;
-
     /**
      * 短信发送回调url
      */

src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/dataobject/sms/SysSmsQueryLogDO.java

@@ -1,15 +1,12 @@
 package cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.sms;
 
-import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
-import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.experimental.Accessors;
 
 import java.io.Serializable;
 import java.util.Date;
-import java.util.List;
 
 /**
  * 短信日志
@@ -28,6 +25,11 @@ public class SysSmsQueryLogDO implements Serializable {
      */
     private Long id;
 
+    /**
+     * 第三方唯一标识
+     */
+    private String apiId;
+
     /**
      * 短信渠道编码(来自枚举类)
      */
@@ -46,8 +48,7 @@ public class SysSmsQueryLogDO implements Serializable {
     /**
      * 手机号
      */
-    @TableField(typeHandler = JacksonTypeHandler.class)
-    private List<String> phones;
+    private String phone;
 
     /**
      * 内容
@@ -66,16 +67,6 @@ public class SysSmsQueryLogDO implements Serializable {
      */
     private Integer gotResult;
 
-    /**
-     * 是否拥有回调函数（0否 1是）
-     */
-    private Integer hadCallback;
-
-    /**
-     * 结果(对象json字符串)
-     */
-    private String sendResultParam;
-
     /**
      * 备注
      */
@@ -91,4 +82,9 @@ public class SysSmsQueryLogDO implements Serializable {
      */
     private Date createTime;
 
+    /**
+     * 发送时间
+     */
+    private Date sendTime;
+
 }

src/main/java/cn/iocoder/dashboard/modules/system/dal/redis/RedisKeyConstants.java

@@ -28,6 +28,6 @@ public interface RedisKeyConstants {
      * key 的 format 的参数是 uuid
      */
     RedisKeyDefine CAPTCHA_CODE = new RedisKeyDefine("captcha_code:%s", STRING, String.class,
-            RedisKeyDefine.TimeoutTypeEnum.DYNAMIC);
+            Duration.ofMinutes(30));
 
 }

src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SmsSendConsumer.java

@@ -1,41 +0,0 @@
-package cn.iocoder.dashboard.modules.system.mq.consumer.sms;
-
-import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
-import cn.iocoder.dashboard.framework.sms.core.SmsResult;
-import cn.iocoder.dashboard.modules.system.mq.message.dept.SysDeptRefreshMessage;
-import cn.iocoder.dashboard.modules.system.mq.message.sms.SmsSendMessage;
-import cn.iocoder.dashboard.modules.system.service.sms.SysSmsChannelService;
-import cn.iocoder.dashboard.modules.system.service.sms.SysSmsQueryLogService;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
-
-/**
- * 针对 {@link SysDeptRefreshMessage} 的消费者
- *
- * @author 芋道源码
- */
-@Component
-@Slf4j
-public class SmsSendConsumer extends AbstractChannelMessageListener<SmsSendMessage> {
-
-    @Resource
-    private SysSmsChannelService smsChannelService;
-
-    @Resource
-    private SysSmsQueryLogService smsQueryLogService;
-
-    @Override
-    public void onMessage(SmsSendMessage message) {
-        log.info("[onMessage][收到 发送短信 消息], content: " + message.toString());
-        AbstractSmsClient smsClient = smsChannelService.getSmsClient(message.getSmsBody().getTemplateCode());
-        String templateApiId = smsChannelService.getSmsTemplateApiIdByCode(message.getSmsBody().getTemplateCode());
-
-        SmsResult result = smsClient.send(templateApiId, message.getSmsBody(), message.getTargetPhones());
-
-        smsQueryLogService.afterSendLog(message.getSmsBody().getSmsLogId(), result);
-    }
-
-}

src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SmsSendMessage.java

@@ -1,25 +0,0 @@
-package cn.iocoder.dashboard.modules.system.mq.message.sms;
-
-import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
-import cn.iocoder.dashboard.framework.sms.core.SmsBody;
-import lombok.Data;
-
-import java.util.Collection;
-import java.util.List;
-
-/**
- * 部门数据刷新 Message
- */
-@Data
-public class SmsSendMessage implements ChannelMessage {
-
-    private SmsBody smsBody;
-
-    private List<String> targetPhones;
-
-    @Override
-    public String getChannel() {
-        return "sms.send";
-    }
-
-}

src/main/java/cn/iocoder/dashboard/modules/system/mq/producer/sms/SmsProducer.java

@@ -1,32 +0,0 @@
-package cn.iocoder.dashboard.modules.system.mq.producer.sms;
-
-import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
-import cn.iocoder.dashboard.framework.sms.core.SmsBody;
-import cn.iocoder.dashboard.modules.system.mq.message.sms.SmsSendMessage;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
-import java.util.List;
-
-/**
- * 短信的 Producer
- */
-@Component
-public class SmsProducer {
-
-    @Resource
-    private StringRedisTemplate stringRedisTemplate;
-
-    /**
-     * 发送 {@link SmsSendMessage} 消息
-     */
-    public void sendSmsSendMessage(SmsBody smsBody, List<String> targetPhoneList) {
-        SmsSendMessage message = new SmsSendMessage();
-        message.setSmsBody(smsBody);
-        message.setTargetPhones(targetPhoneList);
-        // TODO FROM 芋艿 TO ZZF：这块等未来改哈。这个方法目前是广播消费，会导致每个节点都发送一次。等后续封装出 redis stream 消息
-        RedisMessageUtils.sendChannelMessage(stringRedisTemplate, message);
-    }
-
-}

src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/dept/SysDeptRefreshConsumer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/consumer/dept/SysDeptRefreshConsumer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.consumer.dept;
+package cn.iocoder.dashboard.modules.system.redis.mq.consumer.dept;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.dashboard.modules.system.mq.message.dept.SysDeptRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.dept.SysDeptRefreshMessage;
 import cn.iocoder.dashboard.modules.system.service.dept.SysDeptService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/dict/SysDictDataRefreshConsumer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/consumer/dict/SysDictDataRefreshConsumer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.consumer.dict;
+package cn.iocoder.dashboard.modules.system.redis.mq.consumer.dict;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.dashboard.modules.system.mq.message.dict.SysDictDataRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.dict.SysDictDataRefreshMessage;
 import cn.iocoder.dashboard.modules.system.service.dict.SysDictDataService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/permission/SysMenuRefreshConsumer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/consumer/permission/SysMenuRefreshConsumer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.consumer.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.consumer.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.dashboard.modules.system.mq.message.permission.SysMenuRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.permission.SysMenuRefreshMessage;
 import cn.iocoder.dashboard.modules.system.service.permission.SysMenuService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/permission/SysRoleMenuRefreshConsumer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/consumer/permission/SysRoleMenuRefreshConsumer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.consumer.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.consumer.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.dashboard.modules.system.mq.message.permission.SysRoleMenuRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.permission.SysRoleMenuRefreshMessage;
 import cn.iocoder.dashboard.modules.system.service.permission.SysPermissionService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/permission/SysRoleRefreshConsumer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/consumer/permission/SysRoleRefreshConsumer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.consumer.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.consumer.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.dashboard.modules.system.mq.message.permission.SysRoleRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.permission.SysRoleRefreshMessage;
 import cn.iocoder.dashboard.modules.system.service.permission.SysRoleService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/consumer/sms/SmsSendConsumer.java

@@ -0,0 +1,56 @@
+//package cn.iocoder.dashboard.modules.system.redis.mq.consumer.sms;
+//
+//import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener;
+//import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
+//import cn.iocoder.dashboard.framework.sms.core.SmsResult;
+//import cn.iocoder.dashboard.modules.system.redis.mq.message.dept.SysDeptRefreshMessage;
+//import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendMessage;
+//import cn.iocoder.dashboard.modules.system.service.sms.SysSmsChannelService;
+//import cn.iocoder.dashboard.modules.system.service.sms.SysSmsQueryLogService;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.beans.factory.annotation.Autowired;
+//import org.springframework.data.redis.connection.stream.Consumer;
+//import org.springframework.data.redis.connection.stream.ObjectRecord;
+//import org.springframework.data.redis.connection.stream.ReadOffset;
+//import org.springframework.data.redis.connection.stream.StreamOffset;
+//import org.springframework.data.redis.core.StringRedisTemplate;
+//import org.springframework.stereotype.Component;
+//
+//import javax.annotation.Resource;
+//
+///**
+// * 针对 {@link SysDeptRefreshMessage} 的消费者
+// *
+// * @author 芋道源码
+// */
+//@Component
+//@Slf4j
+//public class SmsSendConsumer extends AbstractChannelMessageListener<SmsSendMessage> {
+//
+//    @Resource
+//    private SysSmsChannelService smsChannelService;
+//
+//    @Resource
+//    private SysSmsQueryLogService smsQueryLogService;
+//
+//    @Autowired
+//    StringRedisTemplate redisTemplate;
+//
+//    @Override
+//    public void onMessage(SmsSendMessage message) {
+//
+//        redisTemplate.opsForStream().add(ObjectRecord.create("String", message));
+//
+//        redisTemplate.opsForStream().read(Consumer.from("",""), StreamOffset.create("", ReadOffset.lastConsumed()));
+//
+//
+//
+//        log.info("[onMessage][收到 发送短信 消息], content: " + message.toString());
+//        AbstractSmsClient smsClient = smsChannelService.getSmsClient(message.getSmsBody().getTemplateCode());
+//        String templateApiId = smsChannelService.getSmsTemplateApiIdByCode(message.getSmsBody().getTemplateCode());
+//
+//        SmsResult result = smsClient.send(templateApiId, message.getSmsBody(), message.getTargetPhone());
+//        smsQueryLogService.afterSendLog(message.getSmsBody().getSmsLogId(), result);
+//    }
+//
+//}

src/main/java/cn/iocoder/dashboard/modules/system/mq/message/dept/SysDeptRefreshMessage.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/message/dept/SysDeptRefreshMessage.java

@@ -1,4 +1,4 @@
-package cn.iocoder.dashboard.modules.system.mq.message.dept;
+package cn.iocoder.dashboard.modules.system.redis.mq.message.dept;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
 import lombok.Data;

src/main/java/cn/iocoder/dashboard/modules/system/mq/message/dict/SysDictDataRefreshMessage.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/message/dict/SysDictDataRefreshMessage.java

@@ -1,4 +1,4 @@
-package cn.iocoder.dashboard.modules.system.mq.message.dict;
+package cn.iocoder.dashboard.modules.system.redis.mq.message.dict;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
 import lombok.Data;

src/main/java/cn/iocoder/dashboard/modules/system/mq/message/permission/SysMenuRefreshMessage.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/message/permission/SysMenuRefreshMessage.java

@@ -1,4 +1,4 @@
-package cn.iocoder.dashboard.modules.system.mq.message.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.message.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
 import lombok.Data;

src/main/java/cn/iocoder/dashboard/modules/system/mq/message/permission/SysRoleMenuRefreshMessage.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/message/permission/SysRoleMenuRefreshMessage.java

@@ -1,4 +1,4 @@
-package cn.iocoder.dashboard.modules.system.mq.message.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.message.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
 import lombok.Data;

src/main/java/cn/iocoder/dashboard/modules/system/mq/message/permission/SysRoleRefreshMessage.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/message/permission/SysRoleRefreshMessage.java

@@ -1,4 +1,4 @@
-package cn.iocoder.dashboard.modules.system.mq.message.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.message.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
 import lombok.Data;

src/main/java/cn/iocoder/dashboard/modules/system/mq/producer/dept/SysDeptProducer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/producer/dept/SysDeptProducer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.producer.dept;
+package cn.iocoder.dashboard.modules.system.redis.mq.producer.dept;
 
 import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
-import cn.iocoder.dashboard.modules.system.mq.message.dept.SysDeptRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.dept.SysDeptRefreshMessage;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 

src/main/java/cn/iocoder/dashboard/modules/system/mq/producer/dict/SysDictDataProducer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/producer/dict/SysDictDataProducer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.producer.dict;
+package cn.iocoder.dashboard.modules.system.redis.mq.producer.dict;
 
 import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
-import cn.iocoder.dashboard.modules.system.mq.message.dict.SysDictDataRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.dict.SysDictDataRefreshMessage;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 

src/main/java/cn/iocoder/dashboard/modules/system/mq/producer/permission/SysMenuProducer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/producer/permission/SysMenuProducer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.producer.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.producer.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
-import cn.iocoder.dashboard.modules.system.mq.message.permission.SysMenuRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.permission.SysMenuRefreshMessage;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 

src/main/java/cn/iocoder/dashboard/modules/system/mq/producer/permission/SysPermissionProducer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/producer/permission/SysPermissionProducer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.producer.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.producer.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
-import cn.iocoder.dashboard.modules.system.mq.message.permission.SysRoleMenuRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.permission.SysRoleMenuRefreshMessage;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 

src/main/java/cn/iocoder/dashboard/modules/system/mq/producer/permission/SysRoleProducer.java → src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/producer/permission/SysRoleProducer.java

@@ -1,7 +1,7 @@
-package cn.iocoder.dashboard.modules.system.mq.producer.permission;
+package cn.iocoder.dashboard.modules.system.redis.mq.producer.permission;
 
 import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
-import cn.iocoder.dashboard.modules.system.mq.message.permission.SysRoleRefreshMessage;
+import cn.iocoder.dashboard.modules.system.redis.mq.message.permission.SysRoleRefreshMessage;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 

src/main/java/cn/iocoder/dashboard/modules/system/redis/mq/producer/sms/SmsProducer.java

@@ -0,0 +1,31 @@
+//package cn.iocoder.dashboard.modules.system.redis.mq.producer.sms;
+//
+//import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
+//import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+//import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendMessage;
+//import org.springframework.data.redis.core.StringRedisTemplate;
+//import org.springframework.stereotype.Component;
+//
+//import javax.annotation.Resource;
+//
+///**
+// * 短信的 Producer
+// */
+//@Component
+//public class SmsProducer {
+//
+//    @Resource
+//    private StringRedisTemplate stringRedisTemplate;
+//
+//    /**
+//     * 发送 {@link SmsSendMessage} 消息
+//     */
+//    public void sendSmsSendMessage(SmsBody smsBody, String targetPhone) {
+//        SmsSendMessage message = new SmsSendMessage();
+//        message.setSmsBody(smsBody);
+//        message.setTargetPhone(targetPhone);
+//        // TODO FROM 芋艿 TO ZZF：这块等未来改哈。这个方法目前是广播消费，会导致每个节点都发送一次。等后续封装出 redis stream 消息
+//        RedisMessageUtils.sendChannelMessage(stringRedisTemplate, message);
+//    }
+//
+//}

src/main/java/cn/iocoder/dashboard/modules/system/redis/stream/StreamConsumerRunner.java

@@ -0,0 +1,93 @@
+package cn.iocoder.dashboard.modules.system.redis.stream;
+
+import cn.iocoder.dashboard.framework.redis.core.util.RedisStreamUtils;
+import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendMessage;
+import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendStreamConsumer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ErrorHandler;
+
+import javax.annotation.Resource;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.time.Duration;
+
+
+@Slf4j
+@Component
+public class StreamConsumerRunner implements ApplicationRunner, DisposableBean {
+
+    @Resource
+    RedisConnectionFactory redisConnectionFactory;
+
+    @Resource
+    ThreadPoolTaskExecutor threadPoolTaskExecutor;
+
+    @Resource
+    SmsSendStreamConsumer streamMessageListener;
+
+    @Resource
+    StringRedisTemplate stringRedisTemplate;
+
+    private StreamMessageListenerContainer<String, ObjectRecord<String, SmsSendMessage>> streamMessageListenerContainer;
+
+    @Override
+    public void run(ApplicationArguments args) throws UnknownHostException {
+
+        StreamInfo.XInfoGroups groups = stringRedisTemplate.opsForStream().groups(RedisStreamUtils.KEY_SMS_SEND);
+        if (groups.isEmpty()) {
+            stringRedisTemplate.opsForStream().createGroup(RedisStreamUtils.KEY_SMS_SEND, RedisStreamUtils.GROUP_SMS_SEND);
+        }
+
+
+        // 创建配置对象
+        StreamMessageListenerContainerOptions<String, ObjectRecord<String, SmsSendMessage>> streamMessageListenerContainerOptions = StreamMessageListenerContainerOptions
+                .builder()
+                // 一次性最多拉取多少条消息
+                .batchSize(10)
+                // 执行消息轮询的执行器
+                .executor(this.threadPoolTaskExecutor)
+                // 消息消费异常的handler
+                .errorHandler(new ErrorHandler() {
+                    @Override
+                    public void handleError(Throwable t) {
+                        // throw new RuntimeException(t);
+                        t.printStackTrace();
+                    }
+                })
+                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
+                .pollTimeout(Duration.ZERO)
+                // 序列化器
+                .serializer(new StringRedisSerializer())
+                .targetType(SmsSendMessage.class)
+                .build();
+
+        // 根据配置对象创建监听容器对象
+        StreamMessageListenerContainer<String, ObjectRecord<String, SmsSendMessage>> streamMessageListenerContainer = StreamMessageListenerContainer
+                .create(this.redisConnectionFactory, streamMessageListenerContainerOptions);
+
+        // 使用监听容器对象开始监听消费（使用的是手动确认方式）
+        streamMessageListenerContainer.receive(Consumer.from(RedisStreamUtils.GROUP_SMS_SEND, InetAddress.getLocalHost().getHostName()),
+                StreamOffset.create(RedisStreamUtils.KEY_SMS_SEND, ReadOffset.lastConsumed()), this.streamMessageListener);
+
+        this.streamMessageListenerContainer = streamMessageListenerContainer;
+        // 启动监听
+        this.streamMessageListenerContainer.start();
+
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        this.streamMessageListenerContainer.stop();
+    }
+}

src/main/java/cn/iocoder/dashboard/modules/system/redis/stream/sms/SmsSendMessage.java

@@ -0,0 +1,16 @@
+package cn.iocoder.dashboard.modules.system.redis.stream.sms;
+
+import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+import lombok.Data;
+
+/**
+ * 部门数据刷新 Message
+ */
+@Data
+public class SmsSendMessage {
+
+    private SmsBody smsBody;
+
+    private String targetPhone;
+
+}

src/main/java/cn/iocoder/dashboard/modules/system/redis/stream/sms/SmsSendStreamConsumer.java

@@ -0,0 +1,43 @@
+package cn.iocoder.dashboard.modules.system.redis.stream.sms;
+
+import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
+import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+import cn.iocoder.dashboard.framework.sms.core.SmsResult;
+import cn.iocoder.dashboard.modules.system.service.sms.SysSmsChannelService;
+import cn.iocoder.dashboard.modules.system.service.sms.SysSmsQueryLogService;
+import cn.iocoder.dashboard.util.json.JsonUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.stream.ObjectRecord;
+import org.springframework.data.redis.stream.StreamListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 短信发送流消息监听器
+ *
+ * @author zzf
+ * @date 2021/3/9 16:35
+ */
+@Slf4j
+@Component
+public class SmsSendStreamConsumer implements StreamListener<String, ObjectRecord<String, SmsSendMessage>> {
+
+    @Resource
+    private SysSmsChannelService smsChannelService;
+
+    @Resource
+    private SysSmsQueryLogService smsQueryLogService;
+
+    @Override
+    public void onMessage(ObjectRecord<String, SmsSendMessage> record) {
+        SmsSendMessage message = record.getValue();
+        SmsBody body = message.getSmsBody();
+        log.info("[onMessage][收到 发送短信 消息], content: " + JsonUtils.toJsonString(body));
+        AbstractSmsClient smsClient = smsChannelService.getSmsClient(body.getTemplateCode());
+        String templateApiId = smsChannelService.getSmsTemplateApiIdByCode(body.getTemplateCode());
+
+        SmsResult result = smsClient.send(templateApiId, body, message.getTargetPhone());
+        smsQueryLogService.afterSendLog(body.getSmsLogId(), result);
+    }
+}

src/main/java/cn/iocoder/dashboard/modules/system/redis/stream/sms/SmsSendStreamProducer.java

@@ -0,0 +1,35 @@
+package cn.iocoder.dashboard.modules.system.redis.stream.sms;
+
+import cn.iocoder.dashboard.framework.redis.core.util.RedisStreamUtils;
+import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 短信发送流消息监听器
+ *
+ * @author zzf
+ * @date 2021/3/9 16:35
+ */
+@Slf4j
+@Component
+public class SmsSendStreamProducer {
+
+    @Resource
+    private StringRedisTemplate stringRedisTemplate;
+
+    /**
+     * 发送 {@link SmsSendMessage} 消息
+     */
+    public void sendSmsSendMessage(SmsBody smsBody, String targetPhone) {
+        SmsSendMessage message = new SmsSendMessage();
+        message.setSmsBody(smsBody);
+        message.setTargetPhone(targetPhone);
+
+        RedisStreamUtils.sendChannelMessage(stringRedisTemplate, message);
+    }
+
+}

src/main/java/cn/iocoder/dashboard/modules/system/service/dept/impl/SysDeptServiceImpl.java

@@ -11,7 +11,7 @@ import cn.iocoder.dashboard.modules.system.convert.dept.SysDeptConvert;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dao.dept.SysDeptMapper;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.dept.SysDeptDO;
 import cn.iocoder.dashboard.modules.system.enums.dept.DeptIdEnum;
-import cn.iocoder.dashboard.modules.system.mq.producer.dept.SysDeptProducer;
+import cn.iocoder.dashboard.modules.system.redis.mq.producer.dept.SysDeptProducer;
 import cn.iocoder.dashboard.modules.system.service.dept.SysDeptService;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMultimap;

src/main/java/cn/iocoder/dashboard/modules/system/service/dict/impl/SysDictDataServiceImpl.java

@@ -13,7 +13,7 @@ import cn.iocoder.dashboard.modules.system.convert.dict.SysDictDataConvert;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dao.dict.SysDictDataMapper;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.dict.SysDictDataDO;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.dict.SysDictTypeDO;
-import cn.iocoder.dashboard.modules.system.mq.producer.dict.SysDictDataProducer;
+import cn.iocoder.dashboard.modules.system.redis.mq.producer.dict.SysDictDataProducer;
 import cn.iocoder.dashboard.modules.system.service.dict.SysDictDataService;
 import cn.iocoder.dashboard.modules.system.service.dict.SysDictTypeService;
 import com.google.common.collect.ImmutableTable;

src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysMenuServiceImpl.java

@@ -11,7 +11,7 @@ import cn.iocoder.dashboard.modules.system.dal.mysql.dao.permission.SysMenuMappe
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.permission.SysMenuDO;
 import cn.iocoder.dashboard.modules.system.enums.permission.MenuIdEnum;
 import cn.iocoder.dashboard.modules.system.enums.permission.MenuTypeEnum;
-import cn.iocoder.dashboard.modules.system.mq.producer.permission.SysMenuProducer;
+import cn.iocoder.dashboard.modules.system.redis.mq.producer.permission.SysMenuProducer;
 import cn.iocoder.dashboard.modules.system.service.permission.SysMenuService;
 import cn.iocoder.dashboard.modules.system.service.permission.SysPermissionService;
 import cn.iocoder.dashboard.util.collection.CollectionUtils;

src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java

@@ -11,7 +11,7 @@ import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.permission.SysMe
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.permission.SysRoleDO;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.permission.SysRoleMenuDO;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.permission.SysUserRoleDO;
-import cn.iocoder.dashboard.modules.system.mq.producer.permission.SysPermissionProducer;
+import cn.iocoder.dashboard.modules.system.redis.mq.producer.permission.SysPermissionProducer;
 import cn.iocoder.dashboard.modules.system.service.permission.SysMenuService;
 import cn.iocoder.dashboard.modules.system.service.permission.SysPermissionService;
 import cn.iocoder.dashboard.modules.system.service.permission.SysRoleService;

src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysRoleServiceImpl.java

@@ -15,7 +15,7 @@ import cn.iocoder.dashboard.modules.system.dal.mysql.dao.permission.SysRoleMappe
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.permission.SysRoleDO;
 import cn.iocoder.dashboard.modules.system.enums.permission.RoleCodeEnum;
 import cn.iocoder.dashboard.modules.system.enums.permission.SysRoleTypeEnum;
-import cn.iocoder.dashboard.modules.system.mq.producer.permission.SysRoleProducer;
+import cn.iocoder.dashboard.modules.system.redis.mq.producer.permission.SysRoleProducer;
 import cn.iocoder.dashboard.modules.system.service.permission.SysPermissionService;
 import cn.iocoder.dashboard.modules.system.service.permission.SysRoleService;
 import com.google.common.collect.ImmutableMap;

src/main/java/cn/iocoder/dashboard/modules/system/service/sms/SysSmsChannelService.java

@@ -2,12 +2,14 @@ package cn.iocoder.dashboard.modules.system.service.sms;
 
 import cn.iocoder.dashboard.common.pojo.PageResult;
 import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
+import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
 import cn.iocoder.dashboard.modules.system.controller.sms.vo.SmsChannelAllVO;
 import cn.iocoder.dashboard.modules.system.controller.sms.vo.req.SmsChannelCreateReqVO;
 import cn.iocoder.dashboard.modules.system.controller.sms.vo.req.SmsChannelPageReqVO;
 import cn.iocoder.dashboard.modules.system.controller.sms.vo.resp.SmsChannelEnumRespVO;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.sms.SysSmsChannelDO;
 
+import javax.servlet.ServletRequest;
 import java.util.List;
 
 /**

src/main/java/cn/iocoder/dashboard/modules/system/service/sms/SysSmsQueryLogService.java

@@ -3,6 +3,7 @@ package cn.iocoder.dashboard.modules.system.service.sms;
 import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
 import cn.iocoder.dashboard.framework.sms.core.SmsBody;
 import cn.iocoder.dashboard.framework.sms.core.SmsResult;
+import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
 
 import java.util.List;
 
@@ -16,16 +17,16 @@ public interface SysSmsQueryLogService {
     /**
      * 发送短信前的日志处理
      *
-     * @param smsBody      短信内容
-     * @param targetPhones 发送对象手机号集合
-     * @param client       短信客户端
+     * @param smsBody     短信内容
+     * @param targetPhone 发送对象手机号
+     * @param client      短信客户端
      * @return 生成的日志id
      */
     // TODO FROM 芋艿 to ZZF: async 是针对发送的方式，对于日志不一定需要关心。这样，短信日志，实际就发送前插入，发送后更新结果.
     //   这里只用于记录状态，毕竟异步可能推送失败，此时日志可记录该状态。
 
     // TODO FROM 芋艿 to ZZF：短信日志，群发的情况，应该是每个手机一条哈。虽然是群发，但是可能部分成功，部分失败；对应到短信平台，实际也是多条。
-    void beforeSendLog(SmsBody smsBody, List<String> targetPhones, AbstractSmsClient client);
+    void beforeSendLog(SmsBody smsBody, String targetPhone, AbstractSmsClient client);
 
     /**
      * 发送消息后的日志处理
@@ -35,4 +36,5 @@ public interface SysSmsQueryLogService {
      */
     void afterSendLog(Long logId, SmsResult result);
 
+    void updateSendLogByResultDetail(SmsResultDetail smsResultDetail);
 }

src/main/java/cn/iocoder/dashboard/modules/system/service/sms/SysSmsService.java

@@ -1,10 +1,12 @@
 package cn.iocoder.dashboard.modules.system.service.sms;
 
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.util.ArrayUtil;
 import cn.iocoder.dashboard.framework.sms.core.SmsBody;
+import cn.iocoder.dashboard.framework.sms.core.enums.SmsChannelEnum;
 
 import javax.servlet.ServletRequest;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 
 /**
@@ -22,7 +24,12 @@ public interface SysSmsService {
      * @param smsBody      消息内容
      * @param targetPhones 发送对象手机号列表
      */
-    void send(SmsBody smsBody, List<String> targetPhones);
+    default void send(SmsBody smsBody, List<String> targetPhones) {
+        if (CollectionUtil.isEmpty(targetPhones)) {
+            return;
+        }
+        targetPhones.forEach(s -> this.send(smsBody, s));
+    }
 
     /**
      * 发送消息
@@ -30,9 +37,7 @@ public interface SysSmsService {
      * @param smsBody     消息内容
      * @param targetPhone 发送对象手机号
      */
-    default void send(SmsBody smsBody, String targetPhone) {
-        send(smsBody, Collections.singletonList(targetPhone));
-    }
+    void send(SmsBody smsBody, String targetPhone);
 
     /**
      * 发送消息
@@ -41,14 +46,18 @@ public interface SysSmsService {
      * @param targetPhones 发送对象手机号数组
      */
     default void send(SmsBody smsBody, String... targetPhones) {
+        if (ArrayUtil.isEmpty(targetPhones)) {
+            return;
+        }
         send(smsBody, Arrays.asList(targetPhones));
     }
 
     /**
      * 处理短信发送回调函数
      *
-     * @param request 请求
+     * @param request        请求
      * @return 响应数据
      */
     Object smsSendCallbackHandle(ServletRequest request);
+
 }

src/main/java/cn/iocoder/dashboard/modules/system/service/sms/impl/SysSmsQueryLogServiceImpl.java

@@ -3,16 +3,15 @@ package cn.iocoder.dashboard.modules.system.service.sms.impl;
 import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
 import cn.iocoder.dashboard.framework.sms.core.SmsBody;
 import cn.iocoder.dashboard.framework.sms.core.SmsResult;
+import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
 import cn.iocoder.dashboard.framework.sms.core.property.SmsChannelProperty;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dao.sms.SysSmsQueryLogMapper;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dataobject.sms.SysSmsQueryLogDO;
 import cn.iocoder.dashboard.modules.system.enums.sms.SmsSendStatusEnum;
 import cn.iocoder.dashboard.modules.system.service.sms.SysSmsQueryLogService;
-import cn.iocoder.dashboard.util.json.JsonUtils;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
-import java.util.List;
 
 /**
  * 短信请求日志服务实现类
@@ -27,14 +26,14 @@ public class SysSmsQueryLogServiceImpl implements SysSmsQueryLogService {
     private SysSmsQueryLogMapper logMapper;
 
     @Override
-    public void beforeSendLog(SmsBody smsBody, List<String> targetPhones, AbstractSmsClient client) {
+    public void beforeSendLog(SmsBody smsBody, String targetPhone, AbstractSmsClient client) {
         SysSmsQueryLogDO smsLog = new SysSmsQueryLogDO();
         SmsChannelProperty property = client.getProperty();
 
         smsLog.setChannelCode(property.getCode())
                 .setChannelId(property.getId())
                 .setTemplateCode(smsBody.getTemplateCode())
-                .setPhones(targetPhones)
+                .setPhone(targetPhone)
                 .setContent(smsBody.getParams().toString());
 
         smsLog.setSendStatus(SmsSendStatusEnum.ASYNC.getStatus());
@@ -46,14 +45,19 @@ public class SysSmsQueryLogServiceImpl implements SysSmsQueryLogService {
     public void afterSendLog(Long logId, SmsResult result) {
         SysSmsQueryLogDO smsLog = new SysSmsQueryLogDO();
         smsLog.setId(logId);
-        if (result.getSuccess()) {
-            smsLog.setSendStatus(SmsSendStatusEnum.QUERY_SUCCESS.getStatus());
-            smsLog.setSendResultParam(result.getSendResultParam());
-        } else {
-            smsLog.setSendStatus(SmsSendStatusEnum.QUERY_FAIL.getStatus());
-            smsLog.setRemark(result.getMessage());
-        }
+        smsLog.setApiId(result.getApiId());
+        smsLog.setSendStatus(SmsSendStatusEnum.QUERY_FAIL.getStatus());
+        smsLog.setRemark(result.getCode() + ": " + result.getMessage());
         logMapper.updateById(smsLog);
     }
 
+    @Override
+    public void updateSendLogByResultDetail(SmsResultDetail smsResultDetail) {
+        SysSmsQueryLogDO queryLogDO = new SysSmsQueryLogDO();
+        queryLogDO.setSendStatus(smsResultDetail.getSendStatus());
+        queryLogDO.setSendTime(smsResultDetail.getSendTime());
+        queryLogDO.setRemark(smsResultDetail.getMessage());
+        logMapper.updateByApiId(queryLogDO, smsResultDetail.getApiId());
+    }
+
 }

src/main/java/cn/iocoder/dashboard/modules/system/service/sms/impl/SysSmsSendLogServiceImpl.java

@@ -2,7 +2,6 @@ package cn.iocoder.dashboard.modules.system.service.sms.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
-import cn.iocoder.dashboard.framework.sms.client.NeedQuerySendResultSmsClient;
 import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dao.sms.SysSmsQueryLogMapper;
 import cn.iocoder.dashboard.modules.system.dal.mysql.dao.sms.SysSmsSendLogMapper;
@@ -62,7 +61,7 @@ public class SysSmsSendLogServiceImpl implements SysSmsSendLogService {
             updateQueryLog.setId(queryLog.getId());
 
             // 只处理实现了获取发送结果方法的短信客户端，理论上这里都是满足条件的，以防万一加个判断。
-            if (smsClient instanceof NeedQuerySendResultSmsClient) {
+            /*if (smsClient instanceof NeedQuerySendResultSmsClient) {
                 //初始化点字段值
                 queryLog2SendLong(insertSendLog, queryLog);
 
@@ -89,7 +88,7 @@ public class SysSmsSendLogServiceImpl implements SysSmsSendLogService {
                 //理论上这里都是满足条件的，以防万一加个判断。
                 updateQueryLog.setSendStatus(SmsSendStatusEnum.QUERY_SEND_FAIL.getStatus());
                 smsQueryLogMapper.updateById(updateQueryLog);
-            }
+            }*/
             updateQueryLog.setSendStatus(SmsSendStatusEnum.SEND_SUCCESS.getStatus());
             updateQueryLog.setRemark(String.format("日志(id = %s)对应的客户端没有继承NeedQuerySendResultSmsClient, 不能获取短信结果。", queryLog.getId()));
             smsQueryLogMapper.updateById(updateQueryLog);

src/main/java/cn/iocoder/dashboard/modules/system/service/sms/impl/SysSmsServiceImpl.java

@@ -2,15 +2,16 @@ package cn.iocoder.dashboard.modules.system.service.sms.impl;
 
 import cn.iocoder.dashboard.framework.sms.client.AbstractSmsClient;
 import cn.iocoder.dashboard.framework.sms.core.SmsBody;
-import cn.iocoder.dashboard.modules.system.mq.producer.sms.SmsProducer;
+import cn.iocoder.dashboard.framework.sms.core.SmsClientFactory;
+import cn.iocoder.dashboard.framework.sms.core.SmsResultDetail;
+import cn.iocoder.dashboard.modules.system.redis.stream.sms.SmsSendStreamProducer;
 import cn.iocoder.dashboard.modules.system.service.sms.SysSmsChannelService;
 import cn.iocoder.dashboard.modules.system.service.sms.SysSmsQueryLogService;
 import cn.iocoder.dashboard.modules.system.service.sms.SysSmsService;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
-import java.util.List;
+import javax.servlet.ServletRequest;
 
 /**
  * 短信日志Service实现类
@@ -28,15 +29,23 @@ public class SysSmsServiceImpl implements SysSmsService {
     private SysSmsQueryLogService logService;
 
     @Resource
-    private SmsProducer smsProducer;
+    private SmsSendStreamProducer smsProducer;
+
+    @Resource
+    private SmsClientFactory smsClientFactory;
 
     @Override
-    public void send(SmsBody smsBody, List<String> targetPhones) {
+    public void send(SmsBody smsBody, String targetPhone) {
         AbstractSmsClient client = channelService.getSmsClient(smsBody.getTemplateCode());
-        logService.beforeSendLog(smsBody, targetPhones, client);
-        smsProducer.sendSmsSendMessage(smsBody, targetPhones);
+        logService.beforeSendLog(smsBody, targetPhone, client);
+        smsProducer.sendSmsSendMessage(smsBody, targetPhone);
     }
 
-    // TODO FROM 芋艿 to ZZF：可能要讨论下，对于短信发送来说，貌似只提供异步发送即可。对于业务来说，一定不能依赖短信的发送结果.
+    @Override
+    public Object smsSendCallbackHandle(ServletRequest request) {
+        SmsResultDetail smsResultDetail = smsClientFactory.getSmsResultDetailFromCallbackQuery(request);
+        logService.updateSendLogByResultDetail(smsResultDetail);
+        return smsResultDetail.getCallbackResponseBody();
+    }
 
 }