Spring AMQP

Spring AMQP 提供 RabbitTemplate、@RabbitListener 与声明式 @Bean Exchange/Queue，风格接近 redis-spring 与 kafka-spring。在 spring-boot-essentials 项目加 实验 Profile 即可，主 API 仍 REST + MySQL。

学习目标

• 添加 spring-boot-starter-amqp 并配置连接
• 用 RabbitTemplate 发送、@RabbitListener 消费
• 声明 Exchange、Queue、Binding 与 JSON 转换

核心内容

依赖与配置

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

spring.rabbitmq.host: localhost
spring.rabbitmq.listener.simple.acknowledge-mode: manual
spring.rabbitmq.listener.simple.prefetch: 10

acknowledge-mode: manual 对应 AMQP manual ack；生产环境配合幂等消费（rmq-reliability）。

发送

@Service
@RequiredArgsConstructor
public class OrderNotifyPublisher {
    private final RabbitTemplate rabbitTemplate;

    public void publishPaid(long orderId) {
        rabbitTemplate.convertAndSend(
            "coderoadmap.topic", "order.paid",
            Map.of("orderId", orderId, "at", Instant.now().toString()));
    }
}

在 sb-transaction 事务 提交后 调用，避免双写不一致。

消费

@Component
@Slf4j
public class OrderNotifyListener {

    @RabbitListener(queues = "coderoadmap.notify.email")
    public void onEmail(Map<String, Object> payload, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        log.info("send email for order {}", payload.get("orderId"));
        channel.basicAck(tag, false);
    }
}

也可用 Acknowledgment 参数（Spring AMQP 2.x+）简化 ack。集成测试无 Broker 时 @Disabled 或 Testcontainers，见 sb-test。

练习

• 配置 Jackson2JsonMessageConverter 发送 POJO
• 写 @RabbitListener 监听 topic order.created
• 对比 @KafkaListener 与 @RabbitListener 的配置项差异

小结

RabbitTemplate + @RabbitListener = Boot 里写任务队列的标准姿势。 下一课讲持久化、prefetch 与死信。