RabbitMQ高级特性

cheniy2025/8/15大约 3 分钟

RabbitMQ高级特性

前置知识

在学习本教程之前，请确保您已经：

掌握RabbitMQ的基础知识
了解Spring AMQP的使用
熟悉Java异步编程
具备基本的问题排查能力

死信队列

1. 什么是死信队列？

死信队列（Dead Letter Queue，DLQ）用于处理无法被正常消费的消息。当一条消息变成死信后，可以被重新发送到另一个交换机，这个交换机就是死信交换机（DLX）。

消息变成死信的情况：

消息被拒绝（reject/nack）并且requeue=false
消息过期（TTL）
队列达到最大长度

2. 配置示例

@Configuration
public class DeadLetterConfig {
    // 业务队列
    @Bean
    public Queue businessQueue() {
        return QueueBuilder.durable("business.queue")
                .withArgument("x-dead-letter-exchange", "dlx.exchange")
                .withArgument("x-dead-letter-routing-key", "dlx.key")
                .build();
    }
    
    // 死信队列
    @Bean
    public Queue deadLetterQueue() {
        return new Queue("dlx.queue");
    }
    
    // 死信交换机
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("dlx.exchange");
    }
    
    // 死信队列绑定
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder.bind(deadLetterQueue())
                .to(deadLetterExchange())
                .with("dlx.key");
    }
}

延迟队列

1. TTL + 死信队列实现

延迟队列配置示例

@Configuration
public class DelayQueueConfig {
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable("delay.queue")
                .withArgument("x-dead-letter-exchange", "process.exchange")
                .withArgument("x-dead-letter-routing-key", "process.key")
                .withArgument("x-message-ttl", 5000) // 5秒延迟
                .build();
    }
    
    @Bean
    public Queue processQueue() {
        return new Queue("process.queue");
    }
    
    @Bean
    public DirectExchange processExchange() {
        return new DirectExchange("process.exchange");
    }
    
    @Bean
    public Binding processBinding() {
        return BindingBuilder.bind(processQueue())
                .to(processExchange())
                .with("process.key");
    }
}


### 2. 延迟消息发送

```java
@Service
@Slf4j
public class DelayMessageService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendDelayMessage(String message, int delayTime) {
        rabbitTemplate.convertAndSend("delay.exchange", "delay.key", message,
            msg -> {
                msg.getMessageProperties().setExpiration(String.valueOf(delayTime));
                return msg;
            });
        log.info("发送延迟消息：{}, 延迟：{}ms", message, delayTime);
    }
}

消息优先级

1. 队列配置

@Configuration
public class PriorityConfig {
    @Bean
    public Queue priorityQueue() {
        return QueueBuilder.durable("priority.queue")
                .withArgument("x-max-priority", 10) // 最大优先级为10
                .build();
    }
}

2. 发送优先级消息

@Service
@Slf4j
public class PriorityMessageService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendPriorityMessage(String message, int priority) {
        rabbitTemplate.convertAndSend("priority.exchange", "priority.key", message,
            msg -> {
                msg.getMessageProperties().setPriority(priority);
                return msg;
            });
        log.info("发送优先级消息：{}, 优先级：{}", message, priority);
    }
}

消息追踪

1. 消息追踪配置

@Configuration
public class MessageTraceConfig implements RabbitListenerConfigurer {
    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        
        // 启用消息追踪
        factory.setMessageListener(message -> {
            String messageId = message.getMessageProperties().getMessageId();
            String correlationId = message.getMessageProperties().getCorrelationId();
            log.info("消息追踪 - 消息ID：{}, 关联ID：{}", messageId, correlationId);
        });
        
        registrar.setContainerFactory(factory);
    }
}

插件使用

1. 延迟插件

说明

RabbitMQ的延迟插件（rabbitmq_delayed_message_exchange）提供了更好的延迟消息支持。

@Configuration
public class DelayedConfig {
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("delayed.exchange", "x-delayed-message", true, false, args);
    }
    
    @Bean
    public Queue delayedQueue() {
        return new Queue("delayed.queue");
    }
    
    @Bean
    public Binding delayedBinding() {
        return BindingBuilder.bind(delayedQueue())
                .to(delayedExchange())
                .with("delayed.key")
                .noargs();
    }
}