发布与消费
basicPublish、basicConsume、ack/nack。
30 分钟
发布与消费
掌握 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 进程,观察同一队列的消息如何分摊
- 故意抛异常,对比 与 的行为