发布与消费

掌握 basicPublish / basicConsume 与 ack / nack，就完成 RabbitMQ 最小闭环。Manual ack 保证「处理成功才删消息」，是任务队列可靠性的基础（对比 kafka-consumer 的 offset 提交）。

学习目标

• 声明 durable queue 并发送持久化消息
• 编写 Consumer，basicQos 限制 prefetch
• 正确使用 basicAck / basicNack 与 requeue

核心内容

发布端

channel.queueDeclare("coderoadmap.export", true, false, false, null);
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // persistent
    .contentType("text/plain")
    .build();
channel.basicPublish("", "coderoadmap.export", props,
    "lessonId=99".getBytes(StandardCharsets.UTF_8));

deliveryMode=2 表示消息持久化；还需队列 durable，否则 Broker 重启仍可能丢。业务上先写 MySQL（sb-jpa）再发 MQ，避免「消息发出但事务回滚」。

消费端与 ack

channel.basicQos(10); // 最多 10 条未 ack
DeliverCallback callback = (consumerTag, delivery) -> {
    String body = new String(delivery.getBody(), StandardCharsets.UTF_8);
    try {
        processExport(body);
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
};
channel.basicConsume("coderoadmap.export", false, callback, _ -> {});

第二个参数 autoAck=false 即 manual ack。basicNack(..., requeue=true) 会重新入队；无限失败应转 死信（rmq-reliability）。

prefetch 与工作队列

多个 Consumer 订阅同一 Queue 时，Broker 轮询分发。basicQos(1) 适合慢任务（大文件导出）；过大 prefetch 会导致一个 Worker hoard 消息。类似 Kafka 分区分摊，但 RabbitMQ 靠 队列 + ack 而非 consumer group offset。

Redis List BRPOP 也能做简单队列，但无 Exchange 路由与 DLX；复杂任务仍用 RabbitMQ lab。

练习

• 起两个 Consumer 进程，观察同一队列的消息如何分摊
• 故意抛异常，对比 与 的行为