基于Rdisson實作延時佇列，解決支付延時查單、關單方案-有解無憂

基于Rdisson實作延時佇列，解決支付延時查單、關單方案

- 1.需求
- 2.實作方式
- 3.具體代碼

1.需求

下單支付后，支付回呼因部分因素不可達，導致訂單狀態與微信支付狀態不一致，此時需要服務端主動查詢訂單支付狀態，進行更改訂單狀態，

2.實作方式

基于定時任務
每隔30秒啟動一次，找出最近10分鐘內創建并且未支付的訂單，呼叫微信查單介面核實訂單狀態，系統記錄訂單查詢的次數，在n次查詢之后狀態還是未支付成功，則停止后續查詢，并呼叫關單介面關閉訂單，
基于延時佇列
每隔5秒/30秒/1分鐘/3分鐘/5分鐘/10分鐘/30分鐘呼叫查單介面，若查詢付款成功，不再執行佇列任務，最后一次查詢若還是未回傳支付成功狀態，則停止后續查詢，并呼叫《關單介面》關閉訂單，
延時佇列實作方式（這里不做詳細描述，優缺點自行百度，本文采用redis實作）
- delayqueue
- RabbitMQ
- reids

3.具體代碼

新建物體類 DelayQueueJob 與 ScoredSortedItem

/**
 * @Desecription: 延遲任務
 * @Author: yangyu
 * @Date: 2021/9/10 14:35
 */
@Data
public class DelayQueueJob implements Serializable {

    /**
     * 延遲任務的唯一標識，用于檢索任務
     */
    private long jobId;

    /**
     * 任務的執行時間段
     */
    private List<Long> delayTimeList;

    /**
     * 任務的執行次數
     */
    private Integer exCount = 0;

    /**
     * 任務的執行時間單位
     */
    private TimeUnit timeUnit = TimeUnit.SECONDS;

    /**
     * 任務的執行超時時間
     */
    private long timeout;

    /**
     * 訂單編號，根據編號查詢訂單
     */
    private String orderCode;

    /**
     * 任務型別（具體業務型別）
     */
    private Integer topic;

    /**
     * 任務狀態 0：執行 1：結束
     */
    private int jobType = 0;

}

/**
 * @Desecription: 延時任務 桶
 * @Author: yangyu
 * @Date: 2021/9/10 14:55
 */
@Data
@AllArgsConstructor
public class ScoredSortedItem implements Serializable {

    /**
     * 延遲任務的唯一標識
     */
    private long jobId;

    /**
     * 任務的執行時間
     */
    private long delayTime;

}

定義常量

/**
 * @Desecription: 延時佇列常量
 * @Author: yangyu
 * @Date: 2021/9/10 14:44
 */
public class DelayQueueConstant {

    /*
     * 延時任務池
     * */
    public static final String DELAY_QUEUE_JOB_POOL = "delayQueue:delayQueueJobPool:";

    /*
     * 延時桶
     * */
    public static final String DELAY_BUCKET_KEY_PREFIX = "delayQueue:delayBucket:";

    /*
     * 任務的執行時間段
     * */
    public static final List<Long> SLOT_DELAY_TIME = Arrays.asList(5L, 30L, 60L, 180L, 300L, 600L, 1800L);

    /*
     * 任務的執行固定時間
     * */
    public static final List<Long> FIXED_DELAY_TIME = Arrays.asList(1800L);

}

zset有序佇列操作類

/**
 * @Desecription: 以時間為維度的有序佇列zset 操作類，參考redisson_delay_queue_timeout
 * @Author: yangyu
 * @Date: 2021/9/10 14:50
 */
@Component
public class DelayBucket {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * @Desecription: 添加jobId到延遲任務桶中
     * @Param: key
     * @Param: jobId
     * @Param: delayTimeList
     * @Return:
     * @Author: yangyu
     * @Date: 2021/9/10 14:52
     */
    public void addToBucket(String key, Long jobId, List<Long> delayTimeList, TimeUnit timeUnit) {
        RScoredSortedSet<ScoredSortedItem> scoredSortedSet = redissonClient.getScoredSortedSet(DelayQueueConstant.DELAY_BUCKET_KEY_PREFIX + key);
        long millis = System.currentTimeMillis();
        for (int i = 0; i < delayTimeList.size(); i++) {
            ScoredSortedItem scoredSortedItem = new ScoredSortedItem(jobId, millis + timeUnit.toMillis(delayTimeList.get(i)));
            scoredSortedSet.add(scoredSortedItem.getDelayTime(), scoredSortedItem);
        }
    }

    /**
     * @Desecription: 從延遲任務桶中獲取延遲時間最小的 jodId
     * @Param: jobIdKey
     * @Return: ScoredSortedItem
     * @Author: yangyu
     * @Date: 2021/9/10 15:35
     */
    public ScoredSortedItem getFromBucket(String key) {
        RScoredSortedSet<ScoredSortedItem> scoredSortedSet = redissonClient.getScoredSortedSet(DelayQueueConstant.DELAY_BUCKET_KEY_PREFIX + key);
        if (scoredSortedSet.size() == 0) {
            return null;
        }
        return scoredSortedSet.first();
    }

    /**
     * @Desecription: 從延遲任務桶中洗掉 jod
     * @Param: key
     * @Param scoredSortedItem
     * @Return:
     * @Author: yangyu
     * @Date: 2021/9/10 14:52
     */
    public void deleteFormBucket(String key, ScoredSortedItem scoredSortedItem) {
        RScoredSortedSet<ScoredSortedItem> scoredSortedSet = redissonClient.getScoredSortedSet(DelayQueueConstant.DELAY_BUCKET_KEY_PREFIX + key);
        scoredSortedSet.remove(scoredSortedItem);
    }
}

延時任務類

/**
 * @Desecription: 操作延時任務
 * @Author: yangyu
 * @Date: 2021/9/10 14:43
 */
@Component
public class DelayQueueJobPool {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * @Desecription: 添加延時佇列job
     * @Param: delayQueueJob
     * @Author: yangyu
     * @Date: 2021/9/10 14:43
     */
    public void addDelayQueueJob(DelayQueueJob delayQueueJob, String key) {
        RMap<Long, DelayQueueJob> rMap = redissonClient.getMap(DelayQueueConstant.DELAY_QUEUE_JOB_POOL + key);
        rMap.put(delayQueueJob.getJobId(), delayQueueJob);
    }

    /**
     * @Desecription: 洗掉延時佇列job
     * @Param: jobId
     * @Author: yangyu
     * @Date: 2021/9/10 14:46
     */
    public void deleteDelayQueueJob(Long jobId, String key) {
        RMap<Long, DelayQueueJob> rMap = redissonClient.getMap(DelayQueueConstant.DELAY_QUEUE_JOB_POOL + key);
        rMap.remove(jobId);
    }

    /**
     * @Desecription: 查詢延時佇列job
     * @Param: jobId
     * @Return:
     * @Author: yangyu
     * @Date: 2021/9/10 17:04
     */
    public DelayQueueJob getDelayQueueJob(Long jobId, String key) {
        RMap<Long, DelayQueueJob> rMap = redissonClient.getMap(DelayQueueConstant.DELAY_QUEUE_JOB_POOL + key);
        return rMap.get(jobId);
    }

    /**
     * @Desecription: 修改延時佇列job
     * @Param: jobId
     * @Author: yangyu
     * @Date: 2021/9/10 14:46
     */
    /**
     * @Desecription: 描述方法
     * @Param: jobId
     * @Param: jobId
     * @Return:
     * @Author: yangyu
     * @Date: 2021/9/13 11:09
     */
    public void updateDelayQueueJob(DelayQueueJob delayQueueJob, String key) {
        RMap<Long, DelayQueueJob> rMap = redissonClient.getMap(DelayQueueConstant.DELAY_QUEUE_JOB_POOL + key);
        rMap.replace(delayQueueJob.getJobId(), delayQueueJob);
    }
}

封裝api

@Log4j2
@Component
public class RedisDelayedQueue {

   @Autowired
   private DelayQueueJobPool delayQueueJobPool;

   @Autowired
   private DelayBucket delayBucket;

   /**
    * @Desecription: 添加延遲任務到延遲佇列
    * @Param:
    * @Author: yangyu
    * @Date: 2021/9/10 14:41
    */
   public void push(DelayQueueJob delayQueueJob, String simpleName) {
       delayQueueJobPool.addDelayQueueJob(delayQueueJob, simpleName);
       delayBucket.addToBucket(simpleName, delayQueueJob.getJobId(), delayQueueJob.getDelayTimeList(), delayQueueJob.getTimeUnit());
   }

}

佇列事件監聽介面

/**
 * @Desecription: 佇列事件監聽介面
 * @Author: yangyu
 * @Date: 2021/9/10 10:11
 */
public interface RedisDelayedQueueListenerService<T> {

    void invoke(T t);

}

兩個介面實作類，主要分固定時間和延時時間，具體業務代碼就不貼出來了

/**
 *
 * @Desecription: 延時佇列監聽
 * @Author: yangyu
 * @Date: 2021/9/10 14:32
 */
@Log4j2
@Service
public class DelayedQueueSlotTime implements RedisDelayedQueueListenerService<DelayQueueJob> {

    @Autowired
    private WeChatPay chatPay;

    @Autowired
    private CrcxPayDetailService crcxPayDetailService;

    @Autowired
    private CrcxOrderService crcxOrderService;

    @Autowired
    private DelayQueueJobPool delayQueueJobPool;

    @Override
    public void invoke(DelayQueueJob delayQueueJob) {
        log.info("時間段執行：{}", delayQueueJob);
        // topic == 1 查單延時任務
        if (delayQueueJob.getTopic() == 1) {
            // 1.通過訂單編號，呼叫微信查單介面，查詢微信訂單支付狀態，
            try {
                Map<String, String> queryOrder = chatPay.getQueryOrder(delayQueueJob.getOrderCode());
                // 2.若已付款，對比系統訂單狀態，未付款修改訂單狀態，付款不做處理，
                String tradeState = queryOrder.get("tradeState");
                if (tradeState.equalsIgnoreCase("SUCCESS")) {
                    CrcxOrder crcxOrder = new CrcxOrder();
                    crcxOrder.setOrderCode(delayQueueJob.getOrderCode());
                    crcxOrder = crcxOrderService.getCrcxOrderByCode(crcxOrder);
                    if (crcxOrder.getIsPay() == 0) {
                        // 3.通過訂單編號 查 crcx_pay_detail 支付單-明細
                        CrcxPayDetail crcxPayDetail = new CrcxPayDetail();
                        crcxPayDetail.setTradeCode(delayQueueJob.getOrderCode());
                        crcxPayDetail = crcxPayDetailService.getPayDetail(crcxPayDetail);
                        // 4.進行修改訂單狀態
                        int update = crcxOrderService.updateOrderStatus(crcxPayDetail, queryOrder);
                        if (update >= 1) {
                            log.info("延時佇列：修改訂單支付狀態成功");
                        } else {
                            log.error("延時佇列：修改訂單支付狀態失敗");
                        }
                    }
                    // 5.標記任務不需要再進業務邏輯，因為再 zset 中無法通過jobId 或者 訂單編號獲取對應的List
                    delayQueueJob.setJobType(1);
                    delayQueueJobPool.updateDelayQueueJob(delayQueueJob, this.getClass().getSimpleName());
                }
                // 6.非已付款，不做任何處理，繼續執行延時任務，直到執行完畢，或者訂單已付款，
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

/**
*
* @Desecription: 固定時間佇列監聽
* @Author: yangyu
* @Date: 2021/9/10 13:18
*/
@Log4j2
@Service
public class DelayedQueueFixedTime implements RedisDelayedQueueListenerService<DelayQueueJob> {

    @Autowired
    private CrcxOrderService crcxOrderService;

    @Override
    public void invoke(DelayQueueJob delayQueueJob) {
        log.info("固定時間執行...." + delayQueueJob.getJobId());
        if (delayQueueJob != null) {
            // 查詢訂單狀態是否未支付狀態
            CrcxOrder crcxOrder = new CrcxOrder();
            crcxOrder.setOrderCode(delayQueueJob.getOrderCode());
            crcxOrder = crcxOrderService.getCrcxOrderByCode(crcxOrder);
            if (crcxOrder.getIsPay() == 0) {
                // 將待支付的訂單改為已取消(超時未支付)
                crcxOrder.setStatusCode(50);
                int ref = crcxOrderService.orderPaidTimeout(crcxOrder);
                if (ref == 0) throw new CustomException("自動取消訂單失敗");
            }
        }
    }
}

初始化佇列監聽

/**
 * @Desecription: 初始化佇列監聽
 * @Author: yangyu
 * @Date: 2021/9/10 9:31
 */
@Log4j2
@Component
public class RedisDelayedQueueInit implements ApplicationContextAware {

    @Autowired
    private DelayBucket delayBucket;

    @Autowired
    private DelayQueueJobPool delayQueueJobPool;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisDelayedQueueListenerService> map = applicationContext.getBeansOfType(RedisDelayedQueueListenerService.class);
        for (Map.Entry<String, RedisDelayedQueueListenerService> taskEventListenerEntry : map.entrySet()) {
            String listenerName = taskEventListenerEntry.getValue().getClass().getSimpleName();
            startThread(listenerName, taskEventListenerEntry.getValue());
        }
    }

    /**
     * @Desecription: 啟動執行緒獲取佇列
     * @Param: queueName
     * @Param: redisDelayedQueueListenerService 任務回呼監聽
     * @Return:
     * @Author: yangyu
     * @Date: 2021/9/10 11:32
     */
    private <T> void startThread(String queueName, RedisDelayedQueueListenerService redisDelayedQueueListenerService) {
        Thread thread = new Thread(() -> {
            // 有時間誤差，一般在一秒左右
            log.info("啟動監聽佇列執行緒:{}", queueName);
            while (true) {
                try {
                    ScoredSortedItem item = delayBucket.getFromBucket(queueName);
                    // 沒有任務就堵塞
                    if (item == null) {
                        Thread.sleep(1000);
                        continue;
                    }
                    // 延遲時間沒到
                    if (item.getDelayTime() > System.currentTimeMillis()) {
                        Thread.sleep(1000);
                        continue;
                    }
                    DelayQueueJob delayQueueJob = delayQueueJobPool.getDelayQueueJob(item.getJobId(), queueName);
                    // 延遲任務資料不存在
                    if (delayQueueJob == null) {
                        log.info("延遲任務資料不存在·····");
                        delayBucket.deleteFormBucket(queueName, item);
                        Thread.sleep(1000);
                        continue;
                    }
                    // 如果后面任務不需要執行，不走業務邏輯,洗掉正常執行
                    if (delayQueueJob.getJobType() == 0) {
                        new Thread(() -> {
                            DelayQueueJob job = new DelayQueueJob();
                            BeanUtils.copyProperties(delayQueueJob, job);
                            redisDelayedQueueListenerService.invoke(job);
                        }).start();
                    }
                    Integer exCount = delayQueueJob.getExCount();
                    if (exCount == 1) {
                        delayQueueJobPool.deleteDelayQueueJob(delayQueueJob.getJobId(), queueName);
                        delayBucket.deleteFormBucket(queueName, item);
                        continue;
                    }
                    delayQueueJob.setExCount(--exCount);
                    delayQueueJobPool.updateDelayQueueJob(delayQueueJob, queueName);
                    delayBucket.deleteFormBucket(queueName, item);
                } catch (Exception e) {
                    log.error("監聽佇列執行緒錯誤：", e);
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException ex) {
                    }
                }
            }
        });
        thread.setName(queueName);
        thread.start();
    }
}

主要使用到的依賴包

<dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-boot-starter</artifactId>
            <version>3.10.5</version>
</dependency>

至此整個redis延時佇列就實作完成，如果覺得有用，麻煩點個贊，萬分感謝，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/300445.html

標籤：其他

上一篇：親測有用！Centos7 + Zookeeper集群 + Hadoop集群配置

下一篇：【持續更新】一些linux命令積累——字串處理相關