RabbitMq TTL+死信队列 延迟消息问题记录

--1、cd到rabbitmq默认安装位置
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/plugins

--2、通过ftp工具将插件上传到此目录下

--3、开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

--4、重启MQ服务
systemctl restart rabbitmq-server

--1、通过ftp工具将插件上传到Linux服务器的根目录下

--2、拷贝到docker中rabbitmq插件目录下，rabbitmq_delayed_message_exchange-3.9.0.ez(下载包的全名)
docker cp /rabbitmq_delayed_message_exchange-3.9.0.ez 容器ID:/plugins

--3、进入容器
docker exec -it 容器id /bin/bash

--4、查看插件是否存在(确保2中的操作已经将插件拷贝过来了)
cd plugins
ls |grep delay

--5、开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

--6、退出容器
exit

--7、重启MQ服务
docker restart 容器ID

@Configuration
public class DelayRabbitmqConfig {
 
    /**
     * 声明延迟队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return new Queue(QueueConstant.DelayQueue,
                true,false,false);
    }
 
    /**
     * 声明延迟自定义交换机类型
     * @return
     */
    @Bean
    public CustomExchange delayCustomExchange(){
        HashMap<String, Object> args = new HashMap<>();
//        设置 x-delayed-type 为 direct，当然也可以是 topic 等 发送消息时设置消息头 headers 的 x-delay 属性，即延迟时间，如果不设置消息将会立即投递
        args.put("x-delayed-type","direct");
        return new CustomExchange(ExchangeConstant.DelayCustomerExchange,
                "x-delayed-message",true,false,args);
    }
 
    /**
     * 绑定延迟交换机和队列
     * @return
     */
    @Bean
    public Binding delayQueueAndCustomExchange(){
        return BindingBuilder.bind(delayQueue())
                .to(delayCustomExchange()).with(RoutingKeyConstant.DelayCustomerRoutingKey).noargs();
    }
}

    引入依赖：
    xmlns:util="http://www.springframework.org/schema/util"
    http://www.springframework.org/schema/util
    http://www.springframework.org/schema/util/spring-util-4.0.xsd


	<!-- 定义延时队列 -->
	<!-- durable:是否持久化,宕机恢复后会重持久化日志恢复消息队列 -->
	<!-- exclusive: 仅创建者可以使用的私有队列，断开后自动删除 -->
	<!-- auto_delete: 当所有消费客户端连接断开后，是否自动删除队列 -->
	<rabbit:queue id="taskStartQueue" name="taskStartQueue" durable="true" auto-delete="false" exclusive="false"/>

    	<!--定义延时交换机类型-->
	<util:map id="delayExchangeArgs" map-class="java.util.HashMap">
		<entry key="x-delayed-type" value="direct"/>
	</util:map>

    <!-- 自定义交换机 - 延时交换机 ： rabbitMq 第三方插件 rabbitmq_delayed_message_exchange 实现延时队列 -->
    <!--注意：使用convertAndSend().发送消息后不会进入下面定义的队列中，而是由第三方插件管理，可以从交换机 messages delayed中查看消息数-->
    <bean id="taskExchange" name="delayExchange" class="org.springframework.amqp.core.CustomExchange">
        <constructor-arg name="name" value="taskExchange"/>
        <constructor-arg name="type" value="x-delayed-message"/>
        <constructor-arg name="durable" value="true"/>
        <constructor-arg name="autoDelete" value="false"/>
        <constructor-arg name="arguments" ref="delayExchangeArgs"/>
    </bean>

    <util:map id="emptyMap" map-class="java.util.HashMap" />

    <bean id="taskStartBind" class="org.springframework.amqp.core.Binding">
        <constructor-arg name="destination" value="taskStartQueue"/>
        <constructor-arg name="destinationType" value="QUEUE"/>
        <constructor-arg name="exchange" value="taskExchange"/>
        <constructor-arg name="routingKey" value="delay.task.start"/>
        <constructor-arg name="arguments" ref="emptyMap"/>
    </bean>

//消息发送
final MessagePostProcessor messagePostProcessor = new MyMessagePostProcessor(Integer.valueOf(ttl.toString()));
DisTimingPushDto disTimingPushDto = new DisTimingPushDto();
disTimingPushDto.setOrderId(dispense.getOrderId());
disTimingPushDto.setPushTime(disDispense.getPushTime());
rabbitTemplate.convertAndSend(MsgQueueEnum.TIMING_PUSH.getExchangeName(), MsgQueueEnum.TIMING_PUSH.getQueueName(), disTimingPushDto, messagePostProcessor);

//每条消息时间配置
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;

/**
 * 延迟消息处理器 Processor
 * @author king
 * @date 2022年12月28日 11:14
 */
public class MyMessagePostProcessor implements MessagePostProcessor {

    /**
     * 消息延迟时间，单位：毫秒
     */
    private final Integer TTL;

    public MyMessagePostProcessor(final Integer ttl) {
        this.TTL = ttl;
    }


    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setDelay(TTL);
        return message;
    }
}