ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();

    public static final String DELAY_EXCHANGE = "delay.exchange";

    public static final String DELAY_ROUTE_KEY = "delay.routeKey";

    public static final String DELAY_QUEUE = "delay.queue";

    /**
     * 延時交換機
     * @return 延時交換機
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> arguments = new HashMap<>(1);
        arguments.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE,"x-delayed-message",true,false,arguments);
    }

    /**
     * mq已經(jīng)安裝了延時插件使用,否則得使用延時插件
     * @return 延時發(fā)送隊列。
     */
    @Bean
    public Queue delayQueue() {
        return new Queue(DELAY_QUEUE,true,false,false);
    }

    /**
     * 延時綁定區(qū)
     * @return 延時綁定區(qū)
     */
    @Bean
    public Binding delayBind() {
        return BindingBuilder.bind(this.delayQueue()).to(this.delayExchange()).with(DELAY_ROUTE_KEY).noargs();
    }

package com.linkyoyo.bill.mq.impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.linkyoyo.bill.bo.WorkOrderDelaySenMailActionBO;
import com.linkyoyo.bill.config.RabbitMQConfig;
import com.linkyoyo.bill.mq.DelaySenderService;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

/**
 * 延時發(fā)送
 * @author 鄒 [295006967@qq.com]
 * @date 2022/1/4 20:33
 */
@Slf4j
@RequiredArgsConstructor
@AllArgsConstructor
@Service
public class DelaySenderServiceImpl implements DelaySenderService {

    private final RabbitTemplate rabbitTemplate;

    @Override
    @Async
    public void sendMessageByDelay(JSONObject message, Integer time) {
        if(ObjectUtil.isNull(message) || ObjectUtil.isNull(time)) {
            return;
        }
        rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_EXCHANGE, RabbitMQConfig.DELAY_ROUTE_KEY, message, msg -> {
            msg.getMessageProperties().setHeader("x-delay", time * 1000);
            return msg;
        });
        log.info("延時發(fā)送成功:延時周期時間{}毫秒,消息內(nèi)容為{}", time * 1000, message);
    }

    @Override
    public void sendMessageByDelay(WorkOrderDelaySenMailActionBO actionBO) {
        Integer afterSecond = actionBO.getAfterSecond();
        if(ObjectUtil.isNull(afterSecond)) {
            afterSecond = 0;
        }
        this.sendMessageByDelay(JSONUtil.parseObj(actionBO), afterSecond);
    }
}

	private final DelaySenderService delaySenderService;

    @RabbitHandler
    @RabbitListener(queues = RabbitMQConfig.DELAY_QUEUE)
    public void delayConsumer(Message message) {
        //業(yè)務(wù)邏輯
        this.sendMail(workOrderDelaySenMailActionBO);
        // 判斷是否需要遞歸執(zhí)行定時任務(wù)（實際上就是使用生產(chǎn)者再發(fā)一次延時消息，確認下一次消費）
        if(需要進行定時任務(wù)) {
             this.sendDelayMessageToMq(workOrderDelaySenMailActionBO);
        }
        log.info("信息為:{}", message.getBody());
    }