spring-amqp

Spring AMQP（RabbitMQ）- 快速参考

完整参考：有关交换机类型（topic、fanout）、消费者模式（多方法、请求-回复）、错误处理、DLQ 处理和测试，请参阅 advanced.md。

深入了解：使用 mcp__documentation__fetch_docs，并将 technology 设为 rabbitmq，获取全面的文档。

依赖项

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual
        concurrency: 3
        max-concurrency: 10
        prefetch: 10

交换机、队列和绑定设置

@Configuration
public class RabbitMQConfig {

    public static final String EXCHANGE = "orders.exchange";
    public static final String QUEUE = "orders.queue";
    public static final String ROUTING_KEY = "orders.created";

    @Bean
    public DirectExchange ordersExchange() {
        return new DirectExchange(EXCHANGE);
    }

    @Bean
    public Queue ordersQueue() {
        return QueueBuilder.durable(QUEUE)
            .withArgument("x-dead-letter-exchange", "orders.dlx")
            .withArgument("x-dead-letter-routing-key", "dead")
            .build();
    }

    @Bean
    public Binding ordersBinding() {
        return BindingBuilder.bind(ordersQueue()).to(ordersExchange()).with(ROUTING_KEY);
    }

    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

生产者

@Service
@RequiredArgsConstructor
public class OrderProducer {

    private final RabbitTemplate rabbitTemplate;

    public void sendOrder(OrderEvent event) {
        rabbitTemplate.convertAndSend(EXCHANGE, ROUTING_KEY, event);
    }

    public void sendWithHeaders(OrderEvent event, String correlationId) {
        rabbitTemplate.convertAndSend(EXCHANGE, ROUTING_KEY, event, message -> {
            message.getMessageProperties().setCorrelationId(correlationId);
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        });
    }
}

消费者

@Service
public class OrderConsumer {

    @RabbitListener(queues = "orders.queue")
    public void consume(
            @Payload OrderEvent event,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            @Header(AmqpHeaders.CORRELATION_ID) String correlationId,
            Channel channel) throws IOException {

        try {
            processOrder(event);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            channel.basicNack(deliveryTag, false, false); // Send to DLQ
        }
    }
}

最佳实践

应做	不应做
使用手动确认	未处理消息就自动确认
配置 DLX/DLQ	静默丢弃失败的消息
设置消息 TTL	永久保留消息
使用持久化投递	对重要消息使用临时投递
配置预取	使用不受限制的预取

不应使用此技能的情况

• 原生 RabbitMQ - 使用 rabbitmq 技能进行服务配置
• Kafka 集成 - 使用 spring-kafka 技能
• 通用消息传递模式 - 请咨询 messaging-expert

反模式

反模式	问题	解决方案
未处理消息就自动确认	消息丢失	使用手动确认
未配置 DLQ	失败的消息丢失	配置死信交换机
预取不受限制	内存问题	设置合适的预取数量
在监听器中阻塞	线程耗尽	使用异步处理
未设置重试策略	临时故障导致消息丢失	配置带退避的重试

快速故障排查

问题	诊断	修复
连接被拒绝	检查 RabbitMQ 是否正在运行	验证主机/端口
消息未被消费	检查队列绑定	验证路由键
消息位于 DLQ 中	检查 x-death 标头	修复处理错误
消费者未启动	检查 @RabbitListener	验证队列是否存在
序列化错误	检查消息类型	配置 Jackson 转换器

生产环境检查清单

• 已配置连接工厂
• 使用手动确认模式
• 已配置 DLX/DLQ
• 已设置重试策略
• 已配置消息转换器
• 已启用发布者确认
• 已优化预取数量
• 已实现错误处理器
• 已设置消息 TTL
• 已启用监控

参考文档

• Spring AMQP 参考文档