feature(*): 引入腾讯推送sdk

- 引入腾讯推送sdk

feature(*): 引入腾讯推送sdk
- 引入腾讯推送sdk
张志伟
1 parent 59397026
Showing 13 changed files with 349 additions and 297 deletions
fw-hermes-rpc/pom.xml
fw-hermes-server/src/main/java/cn/fw/hermes/HermesServer.kt
fw-hermes-server/src/main/java/cn/fw/hermes/LogExtend.kt
fw-hermes-server/src/main/java/cn/fw/hermes/config/AutoConfiguration.kt
fw-hermes-server/src/main/java/cn/fw/hermes/config/CustomIdIdentifierGenerator.kt
fw-hermes-server/src/main/java/cn/fw/hermes/config/LaunchConfiguration.kt
fw-hermes-server/src/main/java/cn/fw/hermes/config/ScheduleConfig.kt
fw-hermes-server/src/main/java/cn/fw/hermes/task/.gitkeep
fw-hermes-server/src/main/java/cn/fw/hermes/task/MessageManaTask.kt
fw-hermes-server/src/main/java/cn/fw/hermes/task/MessageSaveTask.kt
fw-hermes-server/src/main/java/cn/fw/hermes/task/MessageSendTask.kt
fw-hermes-service/pom.xml
pom.xml
@@ -42,5 +42,9 @@
             <groupId>cn.fw</groupId>
             <artifactId>fw-common-cache</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-core</artifactId>
+        </dependency>
     </dependencies>
-</project>
 \ No newline at end of file
+</project>
-package cn.fw.hermes;
+package cn.fw.hermes
  
-import cn.fw.security.auth.client.EnableAuthClient;
-import org.mybatis.spring.annotation.MapperScan;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.cache.annotation.EnableCaching;
-import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
-import org.springframework.cloud.openfeign.EnableFeignClients;
-import org.springframework.context.annotation.ComponentScan;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.data.redis.repository.configuration.EnableRedisRepositories;
-import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.transaction.annotation.EnableTransactionManagement;
+import cn.fw.security.auth.client.EnableAuthClient
+import org.mybatis.spring.annotation.MapperScan
+import org.springframework.boot.SpringApplication
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.cache.annotation.EnableCaching
+import org.springframework.cloud.client.discovery.EnableDiscoveryClient
+import org.springframework.cloud.openfeign.EnableFeignClients
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+import org.springframework.data.redis.repository.configuration.EnableRedisRepositories
+import org.springframework.scheduling.annotation.EnableScheduling
+import org.springframework.transaction.annotation.EnableTransactionManagement
  
 /**
  * 启动类
@@ -28,10 +28,10 @@ import org.springframework.transaction.annotation.EnableTransactionManagement;
 @Configuration
 @EnableRedisRepositories
 @MapperScan("cn.fw.**.mapper")
-@ComponentScan({"cn.fw.hermes.*"})
-@EnableFeignClients({"cn.fw.**.sdk"})
-public class HermesServer {
-    public static void main(String[] args) {
-        SpringApplication.run(HermesServer.class, args);
-    }
-}
 \ No newline at end of file
+@ComponentScan("cn.fw.hermes.*")
+@EnableFeignClients("cn.fw.**.sdk")
+class HermesServer
+
+fun main(args: Array<String>) {
+    SpringApplication.run(HermesServer::class.java, *args)
+}
+package cn.fw.hermes
+
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+/**
+ * 日志拓展
+ *
+ * @author : kurisu
+ * @version : 1.0
+ * @desc : 日志拓展
+ * @date : 2023-12-16 15:46
+ */
+
+fun <T : Any> T.logger(): Logger {
+    return LoggerFactory.getLogger(this.javaClass)
+}
+
+fun <T : Any> T.logTrace(msg: String) {
+    logger().trace(msg)
+}
+
+fun <T : Any> T.logTrace(msg: String, vararg var2: Any?) {
+    logger().trace(msg, var2)
+}
+
+fun <T : Any> T.logDebug(msg: String) {
+    logger().debug(msg)
+}
+
+fun <T : Any> T.logDebug(msg: String, vararg var2: Any?) {
+    logger().debug(msg, var2)
+}
+
+fun <T : Any> T.logInfo(msg: String) {
+    logger().info(msg)
+}
+
+fun <T : Any> T.logInfo(msg: String, vararg var2: Any?) {
+    logger().info(msg, var2)
+}
+
+fun <T : Any> T.logWarn(msg: String) {
+    logger().warn(msg)
+}
+
+fun <T : Any> T.logWarn(msg: String, vararg var2: Any?) {
+    logger().warn(msg, var2)
+}
+
+fun <T : Any> T.logError(msg: String) {
+    logger().error(msg)
+}
+
+fun <T : Any> T.logError(e: Throwable) {
+    logger().error("", e)
+}
+
+fun <T : Any> T.logError(msg: String, vararg var2: Any?) {
+    logger().error(msg, var2)
+}
+
+fun <T : Any> T.logError(msg: String, e: Throwable) {
+    logger().error(msg, e)
+}
-package cn.fw.hermes.config;
+package cn.fw.hermes.config
  
-import cn.fw.hermes.service.property.MqttProperty;
-import cn.fw.hermes.service.property.SettingProperty;
-import cn.hutool.core.util.IdUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
-import org.eclipse.paho.client.mqttv3.MqttException;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
+import cn.fw.hermes.logInfo
+import cn.fw.hermes.service.property.MqttProperty
+import cn.fw.hermes.service.property.SettingProperty
+import cn.hutool.core.util.IdUtil
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient
+import org.eclipse.paho.client.mqttv3.MqttException
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
+import org.springframework.boot.context.properties.EnableConfigurationProperties
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
  
 /**
  * mqtt配置
  *
  * @author kurisu
  */
-@Slf4j
 @Configuration
-@EnableConfigurationProperties({MqttProperty.class, SettingProperty.class})
-public class AutoConfiguration {
-
-    private final MqttProperty mqttProperty;
-
-    @Autowired
-    public AutoConfiguration(final MqttProperty mqttProperty) {
-        this.mqttProperty = mqttProperty;
-    }
-
-    @Bean
-    public MqttAsyncClient getMqttClient() throws MqttException {
-        String host = mqttProperty.getHost();
-        int port = mqttProperty.getPort();
-        String nextId = IdUtil.simpleUUID();
-        String clientId = String.format("%s-%s", mqttProperty.getClientId(), nextId);
-        String url = String.format("%s:%d", host, port);
-        log.info("mqtt client initialize : {}", url);
-        return new MqttAsyncClient(url, clientId, new MemoryPersistence());
-    }
+@EnableConfigurationProperties(MqttProperty::class, SettingProperty::class)
+class AutoConfiguration(private val mqttProperty: MqttProperty) {
+    @get:Throws(MqttException::class)
+    @get:Bean
+    val mqttClient: MqttAsyncClient
+        get() {
+            val host = mqttProperty.host
+            val port = mqttProperty.port
+            val nextId = IdUtil.simpleUUID()
+            val clientId = String.format("%s-%s", mqttProperty.clientId, nextId)
+            val url = String.format("%s:%d", host, port)
+            logInfo("mqtt client initialize : {}", url)
+            return MqttAsyncClient(url, clientId, MemoryPersistence())
+        }
 }
-package cn.fw.hermes.config;
+package cn.fw.hermes.config
  
-import com.baomidou.mybatisplus.core.incrementer.IdentifierGenerator;
-import com.github.yitter.idgen.YitIdHelper;
-import org.springframework.stereotype.Component;
+import com.baomidou.mybatisplus.core.incrementer.IdentifierGenerator
+import com.github.yitter.idgen.YitIdHelper
+import org.springframework.stereotype.Component
  
 /**
  * 自定义id生成器
@@ -10,9 +10,8 @@ import org.springframework.stereotype.Component;
  * @author kurisu
  */
 @Component
-public class CustomIdIdentifierGenerator implements IdentifierGenerator {
-    @Override
-    public Long nextId(Object entity) {
-        return YitIdHelper.nextId();
+class CustomIdIdentifierGenerator : IdentifierGenerator {
+    override fun nextId(entity: Any): Long {
+        return YitIdHelper.nextId()
     }
 }
-package cn.fw.hermes.config;
+package cn.fw.hermes.config
  
-import cn.fw.hermes.service.biz.CommonBizService;
-import cn.fw.hermes.service.emqx.MqttService;
-import com.github.yitter.contract.IdGeneratorOptions;
-import com.github.yitter.idgen.YitIdHelper;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.CommandLineRunner;
-import org.springframework.stereotype.Component;
+import cn.fw.hermes.logError
+import cn.fw.hermes.service.biz.CommonBizService
+import cn.fw.hermes.service.emqx.MqttService
+import com.github.yitter.contract.IdGeneratorOptions
+import com.github.yitter.idgen.YitIdHelper
+import org.springframework.beans.factory.DisposableBean
+import org.springframework.boot.CommandLineRunner
+import org.springframework.stereotype.Component
  
 /**
  * @author: Kurisu
  * @date: 2021/2/4 9:13
  */
 @Component
-@Slf4j
-public class LaunchConfiguration implements CommandLineRunner, DisposableBean {
-    private final CommonBizService commonBizService;
-    private final MqttService mqttService;
-
-    @Autowired
-    public LaunchConfiguration(final CommonBizService commonBizService,
-                               final MqttService mqttService) {
-        this.commonBizService = commonBizService;
-        this.mqttService = mqttService;
-    }
-
-    @Override
-    public void run(String... args) {
+class LaunchConfiguration(
+    private val commonBizService: CommonBizService,
+    private val mqttService: MqttService
+) : CommandLineRunner, DisposableBean {
+    override fun run(vararg args: String) {
         try {
             //全局初始化设置WorkerId，默认最大2^16-1，可通过调整 WorkerIdBitLength 增加最大值
-            IdGeneratorOptions options = new IdGeneratorOptions();
-            options.WorkerId = 1;
-            options.WorkerIdBitLength = 3;
-            YitIdHelper.setIdGenerator(options);
-            commonBizService.initSysData();
-            mqttService.initialization();
-        } catch (Exception e) {
-            log.error("初始化系统信息失败", e);
+            val options = IdGeneratorOptions()
+            options.WorkerId = 1
+            options.WorkerIdBitLength = 3
+            YitIdHelper.setIdGenerator(options)
+            commonBizService.initSysData()
+            mqttService.initialization()
+        } catch (e: Exception) {
+            logError("初始化系统信息失败", e)
         }
     }
  
-    @Override
-    public void destroy() {
-        mqttService.disconnect();
+    override fun destroy() {
+        mqttService.disconnect()
     }
 }
-package cn.fw.hermes.config;
-
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.scheduling.annotation.EnableAsync;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-
-import java.util.concurrent.ThreadPoolExecutor;
+package cn.fw.hermes.config
  
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.scheduling.annotation.EnableAsync
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
+import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
  
 /**
  * 定时任务线程池
@@ -15,49 +13,53 @@ import java.util.concurrent.ThreadPoolExecutor;
  */
 @Configuration
 @EnableAsync
-public class ScheduleConfig {
+class ScheduleConfig {
+    @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
+    fun taskExecutor(): ThreadPoolTaskExecutor {
+        val executor = ThreadPoolTaskExecutor()
+        executor.corePoolSize = corePoolSize
+        executor.maxPoolSize = maxPoolSize
+        executor.setQueueCapacity(queueCapacity)
+        executor.keepAliveSeconds = keepAliveTime
+        executor.setThreadNamePrefix(threadNamePrefix)
  
-    /*
+        // 线程池对拒绝任务的处理策略
+        // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
+        executor.setRejectedExecutionHandler(CallerRunsPolicy())
+        // 初始化
+        executor.initialize()
+        return executor
+    }
+
+    companion object {
+        /*
      *   默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，
      *   当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
      *   当队列满了，就继续创建线程，当线程数量大于等于maxPoolSize后，开始使用拒绝策略拒绝
      */
+        /**
+         * 核心线程数（默认线程数）
+         */
+        private const val corePoolSize = 2 shl 4
  
-    /**
-     * 核心线程数（默认线程数）
-     */
-    private static final int corePoolSize = 2 << 4;
-    /**
-     * 最大线程数
-     */
-    private static final int maxPoolSize = 2 << 7;
-    /**
-     * 允许线程空闲时间（单位：默认为秒）
-     */
-    private static final int keepAliveTime = 30;
-    /**
-     * 缓冲队列大小
-     */
-    private static final int queueCapacity = 2 << 10;
-    /**
-     * 线程池名前缀
-     */
-    private static final String threadNamePrefix = "Hermes-Task-";
+        /**
+         * 最大线程数
+         */
+        private const val maxPoolSize = 2 shl 7
  
-    @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
-    public ThreadPoolTaskExecutor taskExecutor() {
-        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
-        executor.setCorePoolSize(corePoolSize);
-        executor.setMaxPoolSize(maxPoolSize);
-        executor.setQueueCapacity(queueCapacity);
-        executor.setKeepAliveSeconds(keepAliveTime);
-        executor.setThreadNamePrefix(threadNamePrefix);
+        /**
+         * 允许线程空闲时间（单位：默认为秒）
+         */
+        private const val keepAliveTime = 30
  
-        // 线程池对拒绝任务的处理策略
-        // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
-        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
-        // 初始化
-        executor.initialize();
-        return executor;
+        /**
+         * 缓冲队列大小
+         */
+        private const val queueCapacity = 2 shl 10
+
+        /**
+         * 线程池名前缀
+         */
+        private const val threadNamePrefix = "Hermes-Task-"
     }
 }
-package cn.fw.hermes.task;
+package cn.fw.hermes.task
  
-import cn.fw.hermes.service.biz.MessageManaBizService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Component;
-
-import java.time.LocalDate;
+import cn.fw.hermes.service.biz.MessageManaBizService
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import java.time.LocalDate
  
 /**
  * 历史消息管理定时器
@@ -17,20 +16,13 @@ import java.time.LocalDate;
  * @date : 2023-07-24 15:44
  */
 @Component
-@ConditionalOnProperty(prefix = "task", name = "switch", havingValue = "on")
-public class MessageManaTask {
-    private final MessageManaBizService messageManaBizService;
-
-    @Autowired
-    public MessageManaTask(final MessageManaBizService messageManaBizService) {
-        this.messageManaBizService = messageManaBizService;
-    }
-
+@ConditionalOnProperty(prefix = "task", name = ["switch"], havingValue = "on")
+class MessageManaTask(private val messageManaBizService: MessageManaBizService) {
     /**
      * 定期删除久远的历史消息
      */
     @Scheduled(cron = "0 0 0/2 * * *")
-    public void deleteOldMsg() {
-        messageManaBizService.removeHistoryMsg(LocalDate.now().minusMonths(6L));
+    fun deleteOldMsg() {
+        messageManaBizService.removeHistoryMsg(LocalDate.now().minusMonths(6L))
     }
 }
-package cn.fw.hermes.task;
+package cn.fw.hermes.task
  
-import cn.fw.hermes.common.constant.Constant;
-import cn.fw.hermes.common.utils.StringUtils;
-import cn.fw.hermes.common.utils.ThreadPoolUtil;
-import cn.fw.hermes.domain.db.MessageData;
-import cn.fw.hermes.domain.dto.MessageConditionDTO;
-import cn.fw.hermes.service.biz.MsgPrepareBizService;
-import cn.fw.hermes.service.data.MessageDataService;
-import cn.hutool.json.JSONArray;
-import cn.hutool.json.JSONUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
+import cn.fw.hermes.common.constant.Constant
+import cn.fw.hermes.common.utils.ThreadPoolUtil
+import cn.fw.hermes.domain.db.MessageData
+import cn.fw.hermes.domain.dto.MessageConditionDTO
+import cn.fw.hermes.logError
+import cn.fw.hermes.service.biz.MsgPrepareBizService
+import cn.fw.hermes.service.data.MessageDataService
+import cn.hutool.json.JSONUtil
+import lombok.extern.slf4j.Slf4j
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.data.redis.core.StringRedisTemplate
+import org.springframework.scheduling.annotation.Async
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import org.springframework.util.CollectionUtils
+import java.util.*
+import java.util.concurrent.CompletableFuture
  
 /**
  * 消息储存任务
@@ -32,44 +26,32 @@ import java.util.concurrent.ExecutorService;
  * @desc : 消息储存任务
  * @date : 2023-07-24 15:05
  */
-@Slf4j
 @Component
-@ConditionalOnProperty(prefix = "task", name = "switch", havingValue = "on")
-public class MessageSaveTask {
-    private final StringRedisTemplate stringRedisTemplate;
-    private final MsgPrepareBizService msgPrepareBizService;
-    private final MessageDataService messageDataService;
-
-    @Autowired
-    public MessageSaveTask(final StringRedisTemplate stringRedisTemplate,
-                           final MsgPrepareBizService msgPrepareBizService,
-                           final MessageDataService messageDataService) {
-        this.stringRedisTemplate = stringRedisTemplate;
-        this.msgPrepareBizService = msgPrepareBizService;
-        this.messageDataService = messageDataService;
-    }
-
+@ConditionalOnProperty(prefix = "task", name = ["switch"], havingValue = "on")
+class MessageSaveTask(
+    private val stringRedisTemplate: StringRedisTemplate,
+    private val msgPrepareBizService: MsgPrepareBizService,
+    private val messageDataService: MessageDataService
+) {
     /**
      * 重试保存失败的消息体
      */
     @Scheduled(initialDelay = 1000 * 5, fixedRate = 1000 * 10)
-    public void retrySaveMsgBody() {
-        List<String> failList = new ArrayList<>();
-        String jsonStr;
-        while ((jsonStr = stringRedisTemplate.opsForSet().pop(Constant.MSG_BODY_SAVE_FAILED_CACHE_KEY)) != null) {
-            JSONArray jsonArray = JSONUtil.parseArray(jsonStr);
-            List<MessageData> list = JSONUtil.toList(jsonArray, MessageData.class);
+    fun retrySaveMsgBody() {
+        val failList: MutableList<String> = mutableListOf()
+        var jsonStr: String?
+        while ((stringRedisTemplate.opsForSet().pop(Constant.MSG_BODY_SAVE_FAILED_CACHE_KEY).also { jsonStr = it }) != null) {
+            val jsonArray = JSONUtil.parseArray(jsonStr)
+            val list = JSONUtil.toList(jsonArray, MessageData::class.java)
             try {
-                messageDataService.saveBatch(list);
-            } catch (Exception e) {
-                if (StringUtils.isValid(jsonStr)) {
-                    failList.add(jsonStr);
-                }
+                messageDataService.saveBatch(list)
+            } catch (e: Exception) {
+                jsonStr?.let { failList.add(it) }
             }
         }
         if (!CollectionUtils.isEmpty(failList)) {
-            String[] array = failList.toArray(new String[0]);
-            stringRedisTemplate.opsForSet().add(Constant.MSG_BODY_SAVE_FAILED_CACHE_KEY, array);
+            val array = failList.toTypedArray<String>()
+            stringRedisTemplate.opsForSet().add(Constant.MSG_BODY_SAVE_FAILED_CACHE_KEY, *array)
         }
     }
  
@@ -78,33 +60,36 @@ public class MessageSaveTask {
      */
     @Async("taskExecutor")
     @Scheduled(initialDelay = 1000 * 5, fixedRate = 1000)
-    public void saveMsg() {
-        List<String> failList = new ArrayList<>();
-        List<CompletableFuture<Void>> futureList = new ArrayList<>();
-        String jsonStr;
-        while ((jsonStr = stringRedisTemplate.opsForList().leftPop(Constant.CACHE_PERSISTENT_MSG_CACHE_KEY)) != null) {
-            final MessageConditionDTO dto = JSONUtil.toBean(jsonStr, MessageConditionDTO.class);
+    fun saveMsg() {
+        val failList: MutableList<String> = mutableListOf()
+        val futureList: MutableList<CompletableFuture<Void>> = mutableListOf()
+        var jsonStr: String?
+
+        while ((stringRedisTemplate.opsForList().leftPop(Constant.CACHE_PERSISTENT_MSG_CACHE_KEY).also { jsonStr = it }) != null) {
+            val dto = JSONUtil.toBean(jsonStr, MessageConditionDTO::class.java)
             if (Objects.isNull(dto)) {
-                continue;
+                continue
             }
-            ExecutorService executor = ThreadPoolUtil.getInstance().getExecutor();
-            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+            val executor = ThreadPoolUtil.getInstance().executor
+            val future = CompletableFuture.runAsync({
                 try {
-                    msgPrepareBizService.saveMsg(dto);
-                    stringRedisTemplate.opsForSet().add(Constant.READY_TO_SEND_MSG_CACHE_KEY, String.valueOf(dto.getId()));
-                } catch (Exception e) {
-                    String _jsonStr = JSONUtil.toJsonStr(dto);
-                    if (StringUtils.isValid(_jsonStr)) {
-                        failList.add(_jsonStr);
+                    msgPrepareBizService.saveMsg(dto)
+                    stringRedisTemplate.opsForSet().add(Constant.READY_TO_SEND_MSG_CACHE_KEY, dto.id.toString())
+                } catch (e: Exception) {
+                    jsonStr?.let {
+                        failList.add(it)
                     }
-                    log.error("持久化消息记录失败: [{}]", e.getMessage());
+
+                    logError("持久化消息记录失败: [{}]", e.message)
                 }
-            }, executor);
-            futureList.add(future);
+            }, executor)
+            futureList.add(future)
         }
-        CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
+
+        CompletableFuture.allOf(*futureList.toTypedArray<CompletableFuture<*>>()).join()
         if (!CollectionUtils.isEmpty(failList)) {
-            stringRedisTemplate.opsForList().rightPushAll(Constant.CACHE_PERSISTENT_MSG_CACHE_KEY, failList);
+            stringRedisTemplate.opsForList().rightPushAll(Constant.CACHE_PERSISTENT_MSG_CACHE_KEY, failList)
         }
     }
+
 }
-package cn.fw.hermes.task;
+package cn.fw.hermes.task
  
-import cn.fw.hermes.common.constant.Constant;
-import cn.fw.hermes.common.utils.StringUtils;
-import cn.fw.hermes.domain.db.MessageHistory;
-import cn.fw.hermes.domain.enums.MsgStatusEnum;
-import cn.fw.hermes.service.biz.AppUpdateBizService;
-import cn.fw.hermes.service.biz.MessageSender;
-import cn.fw.hermes.service.data.MessageHistoryService;
-import com.baomidou.mybatisplus.core.toolkit.Wrappers;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
-
-import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.List;
+import cn.fw.hermes.common.constant.Constant
+import cn.fw.hermes.domain.db.MessageHistory
+import cn.fw.hermes.domain.enums.MsgStatusEnum
+import cn.fw.hermes.logError
+import cn.fw.hermes.service.biz.AppUpdateBizService
+import cn.fw.hermes.service.biz.MessageSender
+import cn.fw.hermes.service.data.MessageHistoryService
+import com.baomidou.mybatisplus.core.toolkit.Wrappers
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.data.redis.core.StringRedisTemplate
+import org.springframework.scheduling.annotation.Async
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import org.springframework.util.CollectionUtils
+import java.time.LocalDateTime
  
 /**
  * 消息发送定时器
@@ -29,37 +25,30 @@ import java.util.List;
  * @desc : 消息发送定时器
  * @date : 2023-07-24 15:44
  */
-@Slf4j
 @Component
-@ConditionalOnProperty(prefix = "task", name = "switch", havingValue = "on")
-public class MessageSendTask {
-    private final StringRedisTemplate stringRedisTemplate;
-    private final MessageHistoryService messageHistoryService;
-    private final MessageSender messageSender;
-    private final AppUpdateBizService appUpdateBizService;
-
-    @Autowired
-    public MessageSendTask(final StringRedisTemplate stringRedisTemplate,
-                           final MessageHistoryService messageHistoryService,
-                           final MessageSender messageSender,
-                           final AppUpdateBizService appUpdateBizService) {
-        this.stringRedisTemplate = stringRedisTemplate;
-        this.messageHistoryService = messageHistoryService;
-        this.messageSender = messageSender;
-        this.appUpdateBizService = appUpdateBizService;
-    }
-
+@ConditionalOnProperty(prefix = "task", name = ["switch"], havingValue = "on")
+class MessageSendTask @Autowired constructor(
+    private val stringRedisTemplate: StringRedisTemplate,
+    private val messageHistoryService: MessageHistoryService,
+    private val messageSender: MessageSender,
+    private val appUpdateBizService: AppUpdateBizService
+) {
     @Scheduled(cron = "0/30 * * * * *")
-    public void ready2Send() {
-        List<MessageHistory> list = messageHistoryService.list(Wrappers.<MessageHistory>lambdaQuery()
-            .eq(MessageHistory::getMsgStatus, MsgStatusEnum.WAITING)
-            .ge(MessageHistory::getCreateTime, LocalDateTime.now().minusDays(7L))
-        );
+    fun ready2Send() {
+        val list = messageHistoryService.list(
+            Wrappers.lambdaQuery<MessageHistory>()
+                .eq(MessageHistory::getMsgStatus, MsgStatusEnum.WAITING)
+                .ge(MessageHistory::getCreateTime, LocalDateTime.now().minusDays(7L))
+        )
         if (CollectionUtils.isEmpty(list)) {
-            return;
+            return
         }
-        String[] array = list.stream().map(r -> String.valueOf(r.getId())).distinct().toArray(String[]::new);
-        stringRedisTemplate.opsForSet().add(Constant.READY_TO_SEND_MSG_CACHE_KEY, array);
+
+        val array = list.map {
+            it.id.toString()
+        }.distinct().toTypedArray()
+
+        stringRedisTemplate.opsForSet().add(Constant.READY_TO_SEND_MSG_CACHE_KEY, *array)
     }
  
     /**
@@ -67,33 +56,32 @@ public class MessageSendTask {
      */
     @Async("taskExecutor")
     @Scheduled(cron = "0/2 * * * * *")
-    public void sendMsg() {
-        List<String> failList = new ArrayList<>();
+    fun sendMsg() {
+        val failList: MutableList<String> = mutableListOf()
         // 限流标志 [最多一次性推送100条]
-        int count = 100;
-        String jsonStr;
-        while ((jsonStr = stringRedisTemplate.opsForSet().pop(Constant.READY_TO_SEND_MSG_CACHE_KEY)) != null) {
-            count--;
+        var count = 100
+        var jsonStr: String?
+        while ((stringRedisTemplate.opsForSet().pop(Constant.READY_TO_SEND_MSG_CACHE_KEY).also { jsonStr = it }) != null) {
+            count--
             try {
-                final Long messageId = Long.valueOf(jsonStr);
-                messageSender.sendMsg(messageId);
-            } catch (Exception e) {
-                if (StringUtils.isValid(jsonStr)) {
-                    failList.add(jsonStr);
+                jsonStr?.toLong()?.let {
+                    messageSender.sendMsg(it)
                 }
-                log.error("推送消息失败", e);
+            } catch (e: Exception) {
+                jsonStr?.let { failList.add(it) }
+                logError("推送消息失败", e)
             }
             if (count < 0) {
-                break;
+                break
             }
         }
         if (!CollectionUtils.isEmpty(failList)) {
-            stringRedisTemplate.opsForSet().add(Constant.READY_TO_SEND_MSG_CACHE_KEY, failList.toArray(new String[0]));
+            stringRedisTemplate.opsForSet().add(Constant.READY_TO_SEND_MSG_CACHE_KEY, *failList.toTypedArray<String>())
         }
     }
  
     @Scheduled(cron = "0/20 * * * * *")
-    public void reSendUpdateMsg() {
-        appUpdateBizService.reSendUpdateMsg();
+    fun reSendUpdateMsg() {
+        appUpdateBizService.reSendUpdateMsg()
     }
 }
@@ -61,6 +61,10 @@
             <groupId>com.github.yitter</groupId>
             <artifactId>yitter-idgenerator</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.github.tpnsPush</groupId>
+            <artifactId>xinge</artifactId>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
@@ -47,6 +47,8 @@
         <yitter.idgenerator>1.0.6</yitter.idgenerator>
         <freemarker.version>2.3.32</freemarker.version>
         <mqttv3.version>1.2.2</mqttv3.version>
+        <kotlinx.coroutines.version>1.7.3</kotlinx.coroutines.version>
+        <tpnspush.version>1.2.4.17</tpnspush.version>
     </properties>
  
     <dependencyManagement>
@@ -155,10 +157,36 @@
                 <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
                 <version>${mqttv3.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.jetbrains.kotlinx</groupId>
+                <artifactId>kotlinx-coroutines-core</artifactId>
+                <version>${kotlinx.coroutines.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.github.tpnsPush</groupId>
+                <artifactId>xinge</artifactId>
+                <version>${tpnspush.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
  
     <dependencies>
+        <!-- kotlin -->
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib</artifactId>
+            <version>${kotlin.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-reflect</artifactId>
+            <version>${kotlin.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib-jdk8</artifactId>
+            <version>${kotlin.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
@@ -174,6 +202,10 @@
     <build>
         <plugins>
             <plugin>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>