简单聊聊RabbitMQ中的队头阻塞问题

// Spring Boot 配置示例
@Configuration
public class RabbitConfig {
    // 定义死信交换器
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx.exchange");
    }

    // 定义死信队列
    @Bean
    public Queue dlxQueue() {
        return new Queue("dlx.queue");
    }

    // 绑定死信队列到 DLX
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx.routing.key");
    }

    // 定义普通队列，设置死信交换器和路由键
    @Bean
    public Queue mainQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "dlx.exchange");
        args.put("x-dead-letter-routing-key", "dlx.routing.key");
        // 可选：设置队列级别的 TTL（所有消息统一过期时间）
        args.put("x-message-ttl", 10000); // 10秒
        return new Queue("main.queue", true, false, false, args);
    }

    // 主队列绑定到默认交换器（根据需要调整）
    @Bean
    public Binding mainBinding() {
        return BindingBuilder.bind(mainQueue()).to(new DirectExchange("default.exchange")).with("main.routing.key");
    }
}

// 发送延迟消息（消息级别 TTL）
public void sendDelayedMessage(String message, int delayMs) {
    rabbitTemplate.convertAndSend("default.exchange", "main.routing.key", message, msg -> {
        // 设置消息过期时间（覆盖队列级别的 TTL）
        msg.getMessageProperties().setExpiration(String.valueOf(delayMs));
        return msg;
    });
}

@RabbitListener(queues = "dlx.queue")
public void handleDelayedMessage(String message) {
System.out.println("处理延迟消息: " + message);
}

Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // 设置队列消息过期时间为 60 秒
channel.queueDeclare(queueName, true, false, false, args);

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .deliveryMode(2) // 消息持久化
        .expiration("60000") // 设置消息过期时间为 60 秒
        .build();
channel.basicPublish(exchangeName, routingKey, properties, message.getBytes());