feat(*): 加入协程

- 加入协程

feat(*): 加入协程
- 加入协程
Kurisu
1 parent fa78338a
Showing 7 changed files with 129 additions and 244 deletions
fw-hestia-common/src/main/java/cn/fw/hestia/common/constant/Constant.kt
fw-hestia-common/src/main/java/cn/fw/hestia/common/constant/MessageConstant.java
fw-hestia-rpc/src/main/java/cn/fw/hestia/rpc/AbsBaseRpcService.java
fw-hestia-rpc/src/main/java/cn/fw/hestia/rpc/passport/TemplateMessageService.kt
fw-hestia-rpc/src/main/java/cn/fw/hestia/rpc/passport/dto/TMParam.kt
fw-hestia-server/src/main/java/cn/fw/hestia/server/task/SendMessageTask.kt
fw-hestia-service/src/main/java/cn/fw/hestia/service/buz/MessageCenterBizService.kt
-package cn.fw.hestia.common.constant;
+package cn.fw.hestia.common.constant
  
 /**
  * 常量
@@ -8,7 +8,10 @@ package cn.fw.hestia.common.constant;
  * @desc : 常量
  * @date : 2023-08-10 10:51
  */
-public interface Constant {
-    String MEMBER_INFO_CACHE = "member:info:id";
-    String MEMBER_INFO_MOBILE_CACHE = "member:info:mobile";
-}
+object Constant {
+    const val MEMBER_INFO_CACHE: String = "member:info:id"
+    const val MEMBER_INFO_MOBILE_CACHE: String = "member:info:mobile"
+
+    const val SUCCEED_STR: String = "succeed"
+    const val MAX_FREQUENCY: Int = 5
+}
 \ No newline at end of file
-package cn.fw.hestia.common.constant;
-
-/**
- * @author : kurisu
- * @className : MessageConstant
- * @description : 常量
- * @date: 2021-09-25
- */
-public interface MessageConstant {
-    String SUCCEED_STR = "succeed";
-    int MAX_FREQUENCY = 5;
-}
-package cn.fw.hestia.rpc;
-
-import cn.fw.hestia.common.utils.StringUtils;
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.redis.core.BoundListOperations;
-import org.springframework.data.redis.core.BoundValueOperations;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.lang.NonNull;
-import org.springframework.lang.Nullable;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.concurrent.TimeUnit;
-
-/**
- * @author : kurisu
- * @className : AbsBaseRpcService
- * @description : 公共方法
- * @date: 2020-12-17 14:13
- */
-@Slf4j
-public abstract class AbsBaseRpcService {
-    /**
-     * Redis工具
-     */
-    @Autowired
-    protected StringRedisTemplate redisTemplate;
-
-    /**
-     * 缓存KEY前缀
-     *
-     * @return
-     */
-    protected abstract String getKeyPrefix();
-
-    /**
-     * 从缓存获取对象
-     *
-     * @param key
-     * @param clazz
-     * @param <E>   获取的对象
-     * @return
-     */
-    @Nullable
-    protected <E> E getFromCache(@NonNull final String key, Class<E> clazz) {
-        String cache = getFromCache(key);
-        if (StringUtils.INSTANCE.isEmpty(cache)) {
-            return null;
-        }
-        return JSON.parseObject(cache, clazz);
-    }
-
-    protected String getFromCache(@NonNull final String key) {
-        String json = null;
-        try {
-            BoundValueOperations<String, String> ops = redisTemplate.boundValueOps(key);
-            json = ops.get();
-        } catch (Exception e) {
-            log.error("从缓存获信息失败[{}]", key, e);
-        }
-        return json;
-    }
-
-    protected void setToCache(@NonNull final String key, @NonNull final String value) {
-        setToCache(key, value, 30);
-    }
-
-    /**
-     * 缓存信息
-     * @param key
-     * @param value
-     * @param timeout 缓存时间(秒)
-     */
-    protected void setToCache(@NonNull final String key, @NonNull final String value, long timeout) {
-        try {
-            redisTemplate.opsForValue().set(key, value, timeout, TimeUnit.SECONDS);
-        } catch (Exception e) {
-            log.error("缓存信息失败[{}][{}]", key, value, e);
-        }
-    }
-
-    protected <E> List<E> getListFromCache(final String key, Class<E> clazz) {
-        try {
-            BoundListOperations<String, String> queue = getQueue(key);
-            final long size = Optional.ofNullable(queue.size()).orElse(0L);
-            if (size > 0) {
-                final List<E> dtos = new ArrayList<>();
-                for (long i = 0; i < size; i++) {
-                    final String json = Objects.requireNonNull(queue.index(i));
-                    dtos.add(JSONObject.parseObject(json, clazz));
-                }
-                return dtos;
-            }
-        } catch (Exception e) {
-            log.error("从缓存获取信息失败[{}]", key, e);
-        }
-        return null;
-    }
-
-    protected void setListToCache(@NonNull final String key, @NonNull final String value) {
-        try {
-            getQueue(key).rightPush(value);
-        } catch (Exception e) {
-            log.error("缓存息失败[{}][{}]", key, value, e);
-        }
-    }
-
-    protected BoundListOperations<String, String> getQueue(@NonNull final String key) {
-        return redisTemplate.boundListOps(key);
-    }
-}
-package cn.fw.hestia.rpc.passport;
+package cn.fw.hestia.rpc.passport
  
-import cn.fw.data.base.domain.common.Message;
-import cn.fw.hestia.common.constant.MessageConstant;
-import cn.fw.hestia.common.utils.StringUtils;
-import cn.fw.hestia.rpc.passport.dto.TMParam;
-import cn.fw.passport.sdk.api.WxMpTemplateMessageApi;
-import cn.fw.passport.sdk.api.param.WxMpTempMessageData;
-import cn.fw.passport.sdk.api.param.WxMpTempMessageParam;
-import com.alibaba.fastjson.JSONArray;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-
-import java.util.List;
+import cn.fw.hestia.common.constant.Constant
+import cn.fw.hestia.common.utils.StringUtils.isValid
+import cn.fw.hestia.rpc.passport.dto.TMParam
+import cn.fw.passport.sdk.api.WxMpTemplateMessageApi
+import cn.fw.passport.sdk.api.param.WxMpTempMessageData
+import cn.fw.passport.sdk.api.param.WxMpTempMessageParam
+import com.alibaba.fastjson.JSONArray
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Service
  
 /**
  * passport系统-模板消息服务
- * <p>
+ *
+ *
  * create at 2019-05-15
  *
  * @author kurisu
  */
-@Slf4j
 @Service
-public class TemplateMessageService {
-
-    private final WxMpTemplateMessageApi wxMpTemplateMessageApi;
-
-    public TemplateMessageService(final WxMpTemplateMessageApi wxMpTemplateMessageApi) {
-        this.wxMpTemplateMessageApi = wxMpTemplateMessageApi;
-    }
+class TemplateMessageService(private val wxMpTemplateMessageApi: WxMpTemplateMessageApi) {
+    private val log = LoggerFactory.getLogger(this::class.java)
  
     /**
      * 发送消息通知
      *
      * @param messageParam
      */
-    public String sendTemplateMessage(TMParam messageParam) {
-        log.info("发送通知，SceneToken ：{}", messageParam.getSceneToken());
+    fun sendTemplateMessage(messageParam: TMParam): String {
+        log.info("发送通知，SceneToken ：{}", messageParam.sceneToken)
         try {
-            WxMpTempMessageParam param = new WxMpTempMessageParam();
-            param.setCusId(messageParam.getMemberId());
-            param.setTempId(messageParam.getTemplateCode());
-            if (StringUtils.INSTANCE.isValid(messageParam.getKeywords())) {
-                List<WxMpTempMessageData> keywords = JSONArray.parseArray(messageParam.getKeywords(), WxMpTempMessageData.class);
-                param.setKeyWordList(keywords);
+            val param = WxMpTempMessageParam()
+            param.cusId = messageParam.memberId
+            param.tempId = messageParam.templateCode
+            if (isValid(messageParam.keywords)) {
+                val keywords = JSONArray.parseArray(
+                    messageParam.keywords,
+                    WxMpTempMessageData::class.java
+                )
+                param.keyWordList = keywords
             }
-            String pagePath = getPagePath(messageParam.getPath(), messageParam.getSceneToken());
-            if (StringUtils.INSTANCE.isValid(pagePath)) {
-                param.setPagePath(pagePath);
+            val pagePath = getPagePath(messageParam.path, messageParam.sceneToken)
+            if (isValid(pagePath)) {
+                param.pagePath = pagePath
             }
-            Message<?> msg = wxMpTemplateMessageApi.send(param);
-            if (!msg.isSuccess()) {
-                log.error("【passport系统】发送模板消息失败：{}", msg.getResult());
-                return msg.getResult();
+            val msg = wxMpTemplateMessageApi.send(param)
+            if (!msg.isSuccess) {
+                log.error("【passport系统】发送模板消息失败：{}", msg.result)
+                return msg.result
             }
-            return MessageConstant.SUCCEED_STR;
-        } catch (Exception e) {
-            log.error("发送模板消息失败", e);
-            return "系统调用异常";
+            return Constant.SUCCEED_STR
+        } catch (e: Exception) {
+            log.error("发送模板消息失败", e)
+            return "系统调用异常"
         }
     }
  
  
-    private String getPagePath(String path, Long sceneToken) {
-        if (StringUtils.INSTANCE.isEmpty(path)) {
-            return null;
-        }
-        StringBuilder sb = new StringBuilder(path);
-        if (StringUtils.INSTANCE.isValid(path)) {
-            sb.append("?sceneToken=").append(sceneToken);
+    private fun getPagePath(path: String?, sceneToken: Long?): String? {
+        return if (!path.isNullOrEmpty()) {
+            val sb = StringBuilder(path)
+            if (isValid(sceneToken)) {
+                sb.append("?sceneToken=").append(sceneToken)
+            }
+            sb.toString()
+        } else {
+            null
         }
-        return sb.toString();
     }
 -}
+}
 \ No newline at end of file
-package cn.fw.hestia.rpc.passport.dto;
-
-import lombok.Data;
-import lombok.ToString;
-
-import java.util.Map;
+package cn.fw.hestia.rpc.passport.dto
  
 /**
- * 续保提醒模板消息param
- *
- * 模板消息样例：
  *
- * {{first.DATA}}
- * 变更类型：{{keyword1.DATA}}
- * 变更结果：{{keyword2.DATA}}
- * {{remark.DATA}}
- *
- * @author kurisu
+ * @className : TMParam
+ * @description :
+ * @author : kurisu
+ * @date : 2023-12-21 22:35
+ * @version :
  */
-@Data
-@ToString
-public class TMParam {
+
+data class TMParam(
     /**
      * 会员id
      */
-    private Long memberId;
+    val memberId: Long?,
     /**
      * 标题
      */
-    private String title;
+    val title: String?,
  
-    private String keywords;
+    val keywords: String?,
     /**
      * 备注
      */
-    private String remark;
+    val remark: String?,
  
     /**
      * 如需跳转小程序，则是小程序页面路径
      */
-    private String path;
+    val path: String?,
  
  
-    private Long sceneToken;
+    val sceneToken: Long?,
     /**
      * 模板code
      * 新版对应模板id
      */
-    private String templateCode;
-}
+    val templateCode: String?,
+)
 package cn.fw.hestia.server.task
  
-import cn.fw.hestia.common.constant.MessageConstant
-import cn.fw.hestia.common.utils.ThreadPoolUtil
+import cn.fw.hestia.common.constant.Constant
 import cn.fw.hestia.domain.db.MessageHistory
 import cn.fw.hestia.domain.enums.MessageStateEnum
 import cn.fw.hestia.service.buz.MessageCenterBizService
 import cn.fw.hestia.service.data.MessageHistoryService
 import com.baomidou.mybatisplus.extension.kotlin.KtQueryWrapper
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.data.redis.core.StringRedisTemplate
@@ -15,7 +17,6 @@ import org.springframework.stereotype.Component
 import org.springframework.util.CollectionUtils
 import java.time.LocalDateTime
 import java.util.*
-import java.util.concurrent.CompletableFuture
  
 /**
  * @author : kurisu
@@ -35,21 +36,23 @@ class SendMessageTask(
     /**
      * 发送模板消息
      */
-    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 5)
+    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 3)
     fun sendNotice() {
         val list = messageHistoryService.list(
             KtQueryWrapper(MessageHistory::class.java)
                 .eq(MessageHistory::state, MessageStateEnum.MADA)
                 .eq(MessageHistory::yn, true)
-                .lt(MessageHistory::frequency, MessageConstant.MAX_FREQUENCY)
+                .lt(MessageHistory::frequency, Constant.MAX_FREQUENCY)
                 .lt(MessageHistory::sendTime, Date())
-                .last("limit 1000")
+                .last("limit 500")
         )
-        if (CollectionUtils.isEmpty(list)) {
+        if (list.isNullOrEmpty()) {
             return
         }
-        for (history in list) {
-            CompletableFuture.runAsync { messageCenterBizService.sendMessage(history) }
+        list.forEach {
+            CoroutineScope(Dispatchers.IO).launch {
+                messageCenterBizService.sendMessage(it)
+            }
         }
     }
  
@@ -63,30 +66,36 @@ class SendMessageTask(
         val list = messageHistoryService.list(
             KtQueryWrapper(MessageHistory::class.java)
                 .lt(MessageHistory::createTime, LocalDateTime.now().minusMonths(6L))
-                .last("limit 500")
+                .last("limit 1000")
         )
         if (CollectionUtils.isEmpty(list)) {
             return
         }
-        list.stream().map(MessageHistory::id).forEach { id: Long? -> id?.let { setOps.add(it.toString()) } }
+        list.forEach {
+            it.id?.run {
+                setOps.add(this.toString())
+            }
+        }
     }
  
     /**
      * 清理数据
      */
-    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 15)
+    @Scheduled(initialDelay = 1000 * 15, fixedRate = 1000 * 15)
     fun clearOldData() {
         val key: String = dataClearKey
         val setOps = stringRedisTemplate.boundSetOps(key)
-        var idStr: String?
-        while ((setOps.pop().also { idStr = it }) != null) {
-            val id = idStr?.toLong()
-            id?.let {
-                CompletableFuture.runAsync(
-                    { messageCenterBizService.clearHistory(it) },
-                    ThreadPoolUtil.executor
-                )
+        var counter = 0
+        val removeIds = mutableSetOf<Long>()
+        while (counter < 2000) {
+            val idStr = setOps.pop() ?: break
+            idStr.toLong().run {
+                removeIds.add(this)
             }
+            counter++
+        }
+        if (removeIds.isNotEmpty()) {
+            messageCenterBizService.clearHistoryBatch(removeIds)
         }
     }
 }
 \ No newline at end of file
@@ -3,7 +3,7 @@ package cn.fw.hestia.service.buz
 import cn.fw.common.businessvalidator.Validator
 import cn.fw.common.data.mybatis.pagination.PageData
 import cn.fw.common.page.AppPage
-import cn.fw.hestia.common.constant.MessageConstant
+import cn.fw.hestia.common.constant.Constant
 import cn.fw.hestia.common.utils.DateUtil
 import cn.fw.hestia.common.utils.StringUtils
 import cn.fw.hestia.component.SendMsgProducer
@@ -22,7 +22,6 @@ import cn.fw.hestia.service.data.SendLogService
 import cn.fw.passport.sdk.api.param.WxMpTempMessageData
 import com.alibaba.fastjson.JSON
 import com.alibaba.fastjson.JSONArray
-import com.baomidou.mybatisplus.core.toolkit.Wrappers
 import com.baomidou.mybatisplus.extension.kotlin.KtQueryWrapper
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.cache.annotation.Cacheable
@@ -110,7 +109,7 @@ class MessageCenterBizService(
                 setOps.add(seanceToken)
             }
         }
-        val succeed = MessageConstant.SUCCEED_STR == result
+        val succeed = Constant.SUCCEED_STR == result
         val frequency = history.frequency!! + 1
         history.frequency = frequency
         if (succeed) {
@@ -133,15 +132,29 @@ class MessageCenterBizService(
      * @param id
      */
     @Transactional(rollbackFor = [Exception::class])
-    fun clearHistory(id: Long?) {
+    fun clearHistory(id: Long) {
         sendLogService.remove(
-            Wrappers.lambdaQuery<SendLog>()
+            KtQueryWrapper(SendLog::class.java)
                 .eq(SendLog::messageId, id)
         )
         messageHistoryService.removeById(id)
     }
  
     /**
+     * 批量清理不使用的数据
+     *
+     * @param idList
+     */
+    @Transactional(rollbackFor = [Exception::class])
+    fun clearHistoryBatch(idList: Set<Long>) {
+        sendLogService.remove(
+            KtQueryWrapper(SendLog::class.java)
+                .`in`(SendLog::messageId, idList)
+        )
+        messageHistoryService.removeByIds(idList)
+    }
+
+    /**
      * 撤回消息
      *
      * @param sceneToken
@@ -251,7 +264,7 @@ class MessageCenterBizService(
             val result = saveLog(
                 createTmParam(messageHistory), messageHistory.id
             ) { r: SendLog -> r.manual = true }
-            val succeed = MessageConstant.SUCCEED_STR == result
+            val succeed = Constant.SUCCEED_STR == result
             if (succeed) {
                 messageHistory.state = MessageStateEnum.SUMI
                 sendMsgProducer.send(MessageSendMq(messageHistory.id, messageHistory.memberId, Date()))
@@ -265,7 +278,7 @@ class MessageCenterBizService(
         get() = settingProperty.tempId ?: "无效code"
  
     private fun saveLog(tmParam: TMParam, messageId: Long?, cons: Consumer<SendLog>): String {
-        val result = MessageConstant.SUCCEED_STR
+        val result = Constant.SUCCEED_STR
         val sendLog = SendLog(messageId, Date(), succeed = true, manual = false, result)
         sendLogService.save(sendLog)
         return result
@@ -317,14 +330,16 @@ class MessageCenterBizService(
     }
  
     private fun createTmParam(history: MessageHistory): TMParam {
-        val tmParam = TMParam()
-        tmParam.setSceneToken(history.id)
-        tmParam.setTitle(history.content)
-        tmParam.setRemark(history.remark)
-        tmParam.setPath(history.pagePath)
-        tmParam.setTemplateCode(history.templateCode)
-        tmParam.setMemberId(history.memberId)
-        tmParam.setKeywords(history.keywords)
+        val (id, memberId, templateCode, title, _, keywords, remark, pagePath) = history
+        val tmParam = TMParam(
+            memberId,
+            title,
+            keywords,
+            remark,
+            path = pagePath,
+            sceneToken = id,
+            templateCode
+        )
         return tmParam
     }
 }
 \ No newline at end of file