feat 通知消息功能编写

_doc/Task.md

@@ -18,6 +18,7 @@
 **任务池**
   - [x] 支付SDK编写
   - [x] 接入支付网关的演示项目
+  - [ ] 订单超时支持数据表级别触发
   - [ ] 支付宝关闭支付时支持撤销方式,
   - [ ] 支持转账操作, 通过支付通道专有参数进行实现, 转账时只能单个通道进行操作
   - [ ] 支付成功回调后, 如果订单已超时, 则进入待退款订单中，提示进行退款，或者自动退款

daxpay-single/daxpay-single-gateway/src/main/java/cn/bootx/platform/daxpay/gateway/controller/TestController.java

@@ -32,7 +32,7 @@ public class TestController {
     @GetMapping("/lock1")
 //    @Lock4j(keys = "#name", acquireTimeout = 50)
     public ResResult<String> lock1(String name){
-        LockInfo lock = lockTemplate.lock(name, 10000, 10);
+        LockInfo lock = lockTemplate.lock(name, 10000, 50);
         if (Objects.isNull(lock)){
             throw new BizException("未获取到锁");
         }

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/code/ClientNoticeTypeEnum.java

@@ -0,0 +1,19 @@
+package cn.bootx.platform.daxpay.service.code;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * 消息通知类型
+ * @author xxm
+ * @since 2024/2/20
+ */
+@Getter
+@AllArgsConstructor
+public enum ClientNoticeTypeEnum {
+    PAY("pay", "支付通知"),
+    REFUND("refund", "退款通知");
+
+    private final String type;
+    private final String name;
+}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/notice/dao/ClientNoticeTaskManager.java

@@ -0,0 +1,18 @@
+package cn.bootx.platform.daxpay.service.core.payment.notice.dao;
+
+import cn.bootx.platform.common.mybatisplus.impl.BaseManager;
+import cn.bootx.platform.daxpay.service.core.payment.notice.entity.ClientNoticeTask;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ *
+ * @author xxm
+ * @since 2024/2/20
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ClientNoticeTaskManager extends BaseManager<ClientNoticeTaskMapper, ClientNoticeTask> {
+}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/notice/dao/ClientNoticeTaskMapper.java

@@ -0,0 +1,14 @@
+package cn.bootx.platform.daxpay.service.core.payment.notice.dao;
+
+import cn.bootx.platform.daxpay.service.core.payment.notice.entity.ClientNoticeTask;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ *
+ * @author xxm
+ * @since 2024/2/20
+ */
+@Mapper
+public interface ClientNoticeTaskMapper extends BaseMapper<ClientNoticeTask> {
+}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/notice/entity/ClientNoticeRecord.java

@@ -0,0 +1,17 @@
+package cn.bootx.platform.daxpay.service.core.payment.notice.entity;
+
+import cn.bootx.platform.common.mybatisplus.base.MpCreateEntity;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * 消息通知任务记录表
+ * @author xxm
+ * @since 2024/2/20
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Accessors(chain = true)
+public class ClientNoticeRecord extends MpCreateEntity {
+}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/notice/entity/ClientNoticeTask.java

@@ -0,0 +1,49 @@
+package cn.bootx.platform.daxpay.service.core.payment.notice.entity;
+
+import cn.bootx.platform.common.mybatisplus.base.MpCreateEntity;
+import cn.bootx.table.modify.annotation.DbColumn;
+import cn.bootx.table.modify.annotation.DbTable;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * 消息通知任务
+ * @author xxm
+ * @since 2024/2/20
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Accessors(chain = true)
+@DbTable(comment = "消息通知任务")
+@TableName("tb_client_notice_task")
+public class ClientNoticeTask extends MpCreateEntity {
+
+    /** 本地订单ID */
+    @DbColumn(comment = "本地订单ID")
+    private Long orderId;
+
+    /**
+     * 回调类型
+     * @see
+     */
+    @DbColumn(comment = "回调类型")
+    private String type;
+
+    /** 消息内容 */
+    @DbColumn(comment = "消息内容")
+    private String content;
+
+    /** 是否发送成功 */
+    @DbColumn(comment = "是否发送成功")
+    private boolean success;
+
+    /** 发送次数 */
+    @DbColumn(comment = "发送次数")
+    private Integer sendTimes;
+
+    /** 发送地址 */
+    @DbColumn(comment = "发送地址")
+    private String url;
+}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/notice/service/ClientNoticeService.java

@@ -0,0 +1,122 @@
+package cn.bootx.platform.daxpay.service.core.payment.notice.service;
+
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
+import cn.bootx.platform.common.redis.RedisClient;
+import cn.bootx.platform.common.spring.exception.RetryableException;
+import cn.bootx.platform.daxpay.service.core.payment.notice.dao.ClientNoticeTaskManager;
+import cn.bootx.platform.daxpay.service.core.payment.notice.entity.ClientNoticeTask;
+import cn.hutool.extra.spring.SpringUtil;
+import cn.hutool.http.ContentType;
+import cn.hutool.http.HttpException;
+import cn.hutool.http.HttpResponse;
+import cn.hutool.http.HttpUtil;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.retry.annotation.Retryable;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * 消息通知任务服务
+ * 总共会发起15次通知，通知频率为15s/15s/30s/3m/10m/20m/30m/30m/30m/60m/3h/3h/3h/6h/6h
+ * @author xxm
+ * @since 2024/2/20
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ClientNoticeService {
+
+    private final ClientNoticeTaskManager taskManager;
+
+    private final RedisClient redisClient;
+
+    private final LockTemplate lockTemplate;
+
+    private final String KEY = "client:notice:task";
+
+    /**
+     * 注册消息通知任务, 失败重试3次, 间隔一秒
+     */
+    @Async("bigExecutor")
+    @Retryable(value = RetryableException.class)
+    public void register(){
+        // 将任务写入到任务表中
+
+        // 同时触发一次通知, 如果成功记录结束
+
+        // 不成功更新任务表, 并注册到Redis对应的表中
+    }
+
+    /**
+     * 从redis中执行任务, 通过定时任务触发
+     */
+    public void run(long start, long end){
+        // 查询Redis任务表,获取任务
+        Set<String> taskIds = redisClient.zrangeByScore(KEY, start, end);
+        // 发起一个异步任务,
+        for (String taskId : taskIds) {
+            SpringUtil.getBean(this.getClass()).run(Long.valueOf(taskId));
+        }
+        // 删除Redis中任务
+        redisClient.zremRangeByScore(KEY, start, end);
+    }
+
+    /**
+     * 执行任务
+     */
+    @Async("asyncExecutor")
+    public void run(Long taskId){
+        LocalDateTime now = LocalDateTime.now();
+        // 开启分布式锁
+        LockInfo lock = lockTemplate.lock(KEY + ":" + taskId,2000, 50);
+        if (Objects.isNull(lock)){
+            throw new RepetitiveOperationException("支付同步处理中，请勿重复操作");
+        }
+        // 查询任务, 进行发送
+        ClientNoticeTask task = taskManager.findById(taskId).orElse(null);
+        try {
+            if (Objects.nonNull(task)){
+                HttpResponse execute = HttpUtil.createPost(task.getUrl())
+                        .body(task.getContent(), ContentType.JSON.getValue())
+                        .timeout(5000)
+                        .execute();
+                String body = execute.body();
+                // 如果响应值等于SUCCESS, 说明发送成功
+                if (Objects.equals(body, "SUCCESS")){
+                    this.successHandler(task);
+                } else {
+                    this.failHandler(task,now);
+                }
+            }
+        } catch (HttpException e) {
+            this.failHandler(task, now);
+        } finally {
+            lockTemplate.releaseLock(lock);
+        }
+    }
+
+    /**
+     * 成功处理
+     */
+    public void successHandler(ClientNoticeTask task){
+        // 记录成功并保存
+        task.setSuccess(true);
+    }
+
+    /**
+     * 失败处理
+     */
+    public void failHandler(ClientNoticeTask task, LocalDateTime now){
+        // 次数+1
+
+
+        // 注册任务到redis中
+    }
+}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/pay/service/PayService.java

@@ -71,7 +71,7 @@ public class PayService {
 
         String businessNo = payParam.getBusinessNo();
         // 加锁
-        LockInfo lock = lockTemplate.lock("payment:pay:" + businessNo);
+        LockInfo lock = lockTemplate.lock("payment:pay:" + businessNo,10000,200);
         if (Objects.isNull(lock)){
             throw new RepetitiveOperationException("正在支付中，请勿重复支付");
         }

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/refund/service/PayRefundService.java

@@ -112,7 +112,7 @@ public class PayRefundService {
         ValidationUtil.validateParam(param);
         // ----------------------------- 发起退款操作 --------------------------------------------
         // 加锁
-        LockInfo lock = lockTemplate.lock("payment:refund:" + payOrder.getId());
+        LockInfo lock = lockTemplate.lock("payment:refund:" + payOrder.getId(),10000,200);
         if (Objects.isNull(lock)){
             throw new RepetitiveOperationException("退款处理中，请勿重复操作");
         }

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/sync/service/PayRefundSyncService.java

@@ -79,7 +79,7 @@ public class PayRefundSyncService {
     @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
     public SyncResult syncRefundOrder(PayRefundOrder refundOrder) {
         // 加锁
-        LockInfo lock = lockTemplate.lock("sync:refund:" + refundOrder.getId());
+        LockInfo lock = lockTemplate.lock("sync:refund:" + refundOrder.getId(),10000,200);
         if (Objects.isNull(lock)) {
             throw new RepetitiveOperationException("退款同步处理中，请勿重复操作");
         }

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/sync/service/PaySyncService.java

@@ -89,7 +89,7 @@ public class PaySyncService {
     @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
     public SyncResult syncPayOrder(PayOrder payOrder) {
         // 加锁
-        LockInfo lock = lockTemplate.lock("sync:payment" + payOrder.getId());
+        LockInfo lock = lockTemplate.lock("sync:payment" + payOrder.getId(),10000,200);
         if (Objects.isNull(lock)){
             throw new RepetitiveOperationException("支付同步处理中，请勿重复操作");
         }

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/timeout/task/PayExpiredTimeTask.java

@@ -40,7 +40,7 @@ public class PayExpiredTimeTask {
 //        log.debug("执行超时取消任务....");
         Set<String> expiredKeys = repository.getExpiredKeys(LocalDateTime.now());
         for (String expiredKey : expiredKeys) {
-            LockInfo lock = lockTemplate.lock("payment:expired:" + expiredKey,10000,0);
+            LockInfo lock = lockTemplate.lock("payment:expired:" + expiredKey,10000,200);
             if (Objects.isNull(lock)){
                 log.warn("支付同步处理中，执行下一个");
                 continue;

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/timeout/task/PayWaitOrderSyncTask.java

@@ -33,7 +33,7 @@ public class PayWaitOrderSyncTask {
         // 从超时订单列表中获取到未超时的订单号
         Set<String> keys = repository.getNormalKeysBy30Day();
         for (String key : keys) {
-            LockInfo lock = lockTemplate.lock("payment:sync:" + key,10000,0);
+            LockInfo lock = lockTemplate.lock("payment:sync:" + key,10000,200);
             if (Objects.isNull(lock)){
                 throw new RepetitiveOperationException("支付同步处理中，请勿重复操作");
             }