RabbitMQ 概览

在微服务与分布式系统中，同步调用容易产生耦合高、峰值流量冲击、故障连锁传递等问题。消息中间件通过异步通信模式解耦上下游服务，而 RabbitMQ 是一款基于 AMQP 协议、功能完备的开源消息代理（Broker）

常见的消息队列产品有 RabbitMQ、ActiveMQ、RocketMQ、Kafka。不同 MQ 定位与特性差异较大，选型需要结合业务吞吐量、可靠性、延迟要求综合判断。下面是四款主流中间件核心指标对比：

对比维度	RabbitMQ	ActiveMQ	RocketMQ	Kafka
维护方	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala & Java
协议支持	AMQP、XMPP、SMTP、STOMP	OpenWire、STOMP、REST、XMPP、AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

RabbitMQ 整体架构与核心组件

整体流转链路： Publisher（生产者） → Exchange（交换机） → Queue（队列） → Consumer（消费者）

Publisher 生产者 消息的发送方，将业务消息推送到 RabbitMQ Broker。
Exchange 交换机 接收生产者发来的消息，根据路由规则把消息分发到对应的队列。 RabbitMQ 提供4种交换机类型：

Queue 队列 真正持久存储消息的容器。消息最终落在队列中，等待消费者拉取消费。交换机本身不存储消息，没有绑定队列时消息会直接丢失。
Consumer 消费者 监听队列，获取消息执行业务逻辑；同一个队列可以挂载多个消费者，实现负载均衡、水平扩容。
Broker：完整的 RabbitMQ 服务实例，包含 Exchange、Queue、连接管理等全部能力。
VirtualHost：Broker 内部的逻辑隔离单元，类似命名空间，同一个 Broker 可以创建多个 vhost，不同 vhost 之间交换机、队列完全隔离

flowchart LR P[Publisher 消息生产者] subgraph Broker["RabbitMQ Server Broker"] subgraph VHost["VirtualHost"] E1[exchange 交换机] E2[exchange 交换机] Q1[queue 队列] Q2[queue 队列] Q3[queue 队列] E1 --> Q1 E2 --> Q1 E2 --> Q2 E2 --> Q3 end end C1[consumer 消费者] C2[consumer 消费者] P -.-> E1 P -.-> E2 Q1 -.-> C1 Q1 -.-> C2 Q2 -.-> C1 Q2 -.-> C2 Q3 -.-> C1 Q3 -.-> C2

快速开始

访问地址： http://localhost:15672
账号：admin
密码：123456
关键点：镜像必须带 -management： rabbitmq:3-management —— 内置web管理后台应用的版本

version: '3.8'

services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    restart: always
    ports:
      # AMQP 通信端口（代码连接使用）
      - "5672:5672"
      # Web管理后台端口
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: 123456
    volumes:
      - rabbit_data:/var/lib/rabbitmq

volumes:
  rabbit_data:

给绑定了队列的交换机发一条消息，测试是否联通

在 Queues & Streams 选项下可以找到 get messages获取所有的消息

在 Admin下侧边导航 Virtual Hosts 可以新增除了默认的“/” 以外的虚拟主机

选择创建 Queues或Exchanges时，就可以选择不同的虚拟主机

RabbitMQ Work Queues模型

核心思路

一个队列可以被多个消费者同时监听（competing consumers）。生产者只管把任务丢进队列，不关心谁来处理；RabbitMQ 负责把消息派发给这些消费者，从而实现横向扩容

三个关键机制

机制	作用
`durable: true` + `persistent: true`	队列和消息都持久化，RabbitMQ 重启也不丢任务
`channel.prefetch(1)`	公平分发：处理完当前消息（ack）之前不再收到下一条，避免忙的更忙、闲的更闲
手动 ack（`noAck: false`）	任务真正处理完才确认；worker 中途挂掉，未 ack 的消息会被重新派发给别的消费者

示例一：`task_queue`（公平分发）

生产者只管发任务：

// apps/publish/src/routes/task.route.ts
getRabbitChannel().sendToQueue(TASK_QUEUE_NAME, Buffer.from(body.task), {
  persistent: true,
});

消费者用 prefetch(1) + 手动 ack 实现"谁干得快谁多拿活"：

// apps/consumer/src/workers/task.worker.ts
await channel.prefetch(1);

await channel.consume(TASK_QUEUE_NAME, async (msg) => {
  if (!msg) return;
  await sleep(seconds * 1000); // 模拟耗时任务
  channel.ack(msg);            // 处理完才确认
}, { noAck: false });

配合环境变量（PORT / WORKER_ID）可以起多个 consumer 进程，共同竞争同一个队列，实测处理快的 worker 会多分到任务

示例二：`work.queue`（默认轮询，对比"不公平"的情况）

故意不设置 prefetch，还原 RabbitMQ 默认行为：不管消费者处理能力，严格按 1:1 轮询派发

// apps/consumer/src/workers/work.worker.ts
async function listen(channel, label, messagesPerSecond) {
  const delayMs = 1000 / messagesPerSecond;
  await channel.consume(WORK_QUEUE_NAME, async (msg) => {
    if (!msg) return;
    await sleep(delayMs);   // 模拟这个消费者的处理速度
    channel.ack(msg);
  }, { noAck: false });
}

// 两个消费者：一个快（50条/秒），一个慢（5条/秒）
startWorkConsumers(fastChannel, slowChannel);

两个消费者要用各自独立的 channel（createChannel() 新开一条），因为 prefetch 是以 channel 为作用域的，共用一条会互相影响

实测发 30 条消息后，快、慢消费者各分到 15 条——数量完全一样，跟处理速度无关。慢的那一侧会明显堆积"已投递但未处理完"的消息，直观暴露了默认轮询的问题，对比示例一的 prefetch(1) 更容易理解"公平分发"到底解决了什么

秉承压榨劳动力的原则：

想让处理能力强的消费者多干活 → 一定要设置 channel.prefetch(1) + 手动 ack
不设置 prefetch 时，RabbitMQ 只按消费者数量轮询，不看实际处理速度，容易造成慢消费者堆积。
队列 durable 和消息 persistent 要配套使用，否则任意一边丢失持久化语义就没意义。

交换机类型

RabbitMQ 里交换机（Exchange）才是消息真正的落脚点，队列只是订阅在交换机上的"收件箱",
交换机分为以下几类：Fanout 、direct、topic

Fanout 交换机

Fanout 交换机是最简单的一种交换机类型，语义就是广播：一条消息进来，会被原样复制给所有绑定在这个交换机上的队列，路由过程完全不看 routing key，也不看消息内容

graph LR P["Publisher"] -- "publish" --> E["交换机 hmall.fanout"] E -- "广播" --> Q1["队列 fanout.queue1"] E -- "广播" --> Q2["队列 fanout.queue2"] Q1 --> C1["消费者1"] Q2 --> C2["消费者2"]

和之前模型的区别

模型	生产者调用	队列怎么收到消息
默认交换机（简单发布/订阅、Work Queues）	`sendToQueue(队列名, ...)`	队列名本身就是隐式路由，天然绑定到默认交换机
Fanout 交换机	`publish(交换机名, "", ...)`	必须显式 `bindQueue(队列, 交换机, "")`，交换机才知道要广播给谁

sendToQueue 本质是 publish("", 队列名, ...) 的语法糖，走的就是 RabbitMQ 内置的默认交换机；这次换成自定义的 elysia.fanout 交换机，多了一步绑定，是理解上的关键差异

示例场景

参照下面这张图里的四个步骤来实现：声明两个队列 fanout.queue1、fanout.queue2；声明交换机 elysia.fanout 并把两个队列绑定上去；consumer 服务里分别写两个方法监听这两个队列；publisher 里写一个测试方法，把消息发到 elysia.fanout

consumer 端：声明交换机、队列、绑定、监听

// apps/consumer/src/listeners/fanout.listener.ts
import {
  FANOUT_EXCHANGE_NAME,
  FANOUT_QUEUE_1,
  FANOUT_QUEUE_2,
  type Channel,
} from "@rabbitmq-demo/shared";

export async function startFanoutListeners(channel: Channel) {
  await channel.assertExchange(FANOUT_EXCHANGE_NAME, "fanout", {
    durable: true,
  });

  await channel.assertQueue(FANOUT_QUEUE_1, { durable: true });
  await channel.assertQueue(FANOUT_QUEUE_2, { durable: true });

  await channel.bindQueue(FANOUT_QUEUE_1, FANOUT_EXCHANGE_NAME, "");
  await channel.bindQueue(FANOUT_QUEUE_2, FANOUT_EXCHANGE_NAME, "");

  channel.consume(FANOUT_QUEUE_1, (msg) => {
    if (!msg) return;
    console.log(`📢 [${FANOUT_QUEUE_1}] 收到广播: ${msg.content.toString()}`);
    channel.ack(msg);
  });

  channel.consume(FANOUT_QUEUE_2, (msg) => {
    if (!msg) return;
    console.log(`📢 [${FANOUT_QUEUE_2}] 收到广播: ${msg.content.toString()}`);
    channel.ack(msg);
  });
}

这段代码里有四个动作，缺一个都跑不起来：

assertExchange(名字, "fanout", ...)：声明一个 fanout 类型的交换机，durable: true 让它在 RabbitMQ 重启后依然存在
两次 assertQueue：声明两个持久化队列，跟之前的例子一样
两次 bindQueue(队列, 交换机, routingKey)：这是全新的一步，把队列挂到交换机上；第三个参数传空字符串，因为 fanout 交换机根本不用它做过滤
两次 consume：给两个队列各自注册一个处理函数，日志里带上队列名，方便对照广播效果

交换机、队列的声明和绑定都放在 consumer 这一侧，是因为这三者不存储消息本身，必须先有队列挂好，生产者发消息时才有地方可以广播过去；生产者完全不用参与这个过程

publisher 端：发布消息到交换机

// apps/publish/src/routes/fanout.route.ts
import { Elysia, t } from "elysia";
import { FANOUT_EXCHANGE_NAME, getRabbitChannel } from "@rabbitmq-demo/shared";

export const fanoutRoutes = new Elysia().post(
  "/fanout/publish",
  ({ body }) => {
    getRabbitChannel().publish(
      FANOUT_EXCHANGE_NAME,
      "",
      Buffer.from(body.message),
      { persistent: true },
    );
    return {
      status: "ok",
      exchange: FANOUT_EXCHANGE_NAME,
      message: body.message,
    };
  },
  { body: t.Object({ message: t.String() }) },
);

调用的是 channel.publish，不是 sendToQueue；第一个参数是交换机名，第二个参数 routing key 同样传空字符串，因为发到 fanout 交换机时这个参数没有意义

两个服务启动时都会 assertExchange 一次，这是幂等操作，两边各声明一次互不冲突，目的是保证不管哪个服务先启动，交换机都已经存在

验证结果

用 curl -X POST http://localhost:3000/fanout/publish -d '{"message":"广播消息"}' 发一条消息后，去 RabbitMQ 管理后台看两个队列的统计，fanout.queue1 和 fanout.queue2 各自 ack 了一次，交换机的 publish_in 也是一次，说明同一条消息确实被复制成两份分别投递给了两个队列，而不是只有一个队列能收到

注意：

想让一条消息同时被多个队列收到，需要自定义交换机（这里是 fanout 类型），并且用 bindQueue 把队列挂上去，默认交换机做不到这件事
fanout 交换机完全不看 routing key，绑定了就一定收到，不绑定就一定收不到，这也是它和后面 direct、topic 交换机最大的差异点
生产者只依赖交换机名，队列的声明和绑定是消费者的职责，新增一个消费队列不需要改动生产者代码，这正是发布/订阅模型解耦的地方

Direct 交换机

Direct 是精准匹配路由的交换机，消息投递规则：消息携带的routing key必须完全等于队列绑定时指定的routing key，消息才会路由到该队列。

Fanout是广播不看routingKey；Direct是严格字符串相等匹配routingKey。

适用场景

按消息类型、业务类型精准分发消息，例如日志分级：error、warn、info，不同消费服务只监听对应级别日志。
单条消息只路由到符合key的队列，不是全量广播。
同一个队列可以用多个不同routingKey绑定到direct交换机；多个队列也可以用同一个routingKey绑定，实现同key多队列。

flowchart LR P["Publisher"] -- "publish routingKey=error" --> E["交换机 hmall.direct"] E -- "key=error" --> Q1["队列 error.queue"] E -- "key=warn" --> Q2["队列 warn.queue"] E -- "key=info" --> Q3["队列 info.queue"] Q1 --> C1["消费者：只处理错误日志"] Q2 --> C2["消费者：只处理警告日志"] Q3 --> C3["消费者：只处理普通信息日志"]

核心API区别

Fanout：bindQueue(队列,交换机,"") routingKey填空
Direct：bindQueue(队列,交换机, "error")，绑定的时候指定routingKey；生产者publish时第二个参数填写完全一致字符串。

Consumer端代码示例（声明direct交换机、队列、绑定）

// apps/consumer/src/listeners/direct.listener.ts
import {
  DIRECT_EXCHANGE_NAME,
  DIRECT_QUEUE_ERROR,
  DIRECT_QUEUE_WARN,
  DIRECT_QUEUE_INFO,
  type Channel,
} from "@rabbitmq-demo/shared";

export async function startDirectListeners(channel: Channel) {
  // 声明direct类型交换机，持久化
  await channel.assertExchange(DIRECT_EXCHANGE_NAME, "direct", {
    durable: true,
  });

  // 声明3个持久化队列
  await channel.assertQueue(DIRECT_QUEUE_ERROR, { durable: true });
  await channel.assertQueue(DIRECT_QUEUE_WARN, { durable: true });
  await channel.assertQueue(DIRECT_QUEUE_INFO, { durable: true });

  // 绑定队列 + 指定各自的routingKey
  await channel.bindQueue(DIRECT_QUEUE_ERROR, DIRECT_EXCHANGE_NAME, "error");
  await channel.bindQueue(DIRECT_QUEUE_WARN, DIRECT_EXCHANGE_NAME, "warn");
  await channel.bindQueue(DIRECT_QUEUE_INFO, DIRECT_EXCHANGE_NAME, "info");

  // 监听error队列，只接收routingKey=error消息
  channel.consume(DIRECT_QUEUE_ERROR, (msg) => {
    if (!msg) return;
    console.log(`🔴 [ERROR] ${msg.content.toString()}`);
    channel.ack(msg);
  });

  // 监听warn队列
  channel.consume(DIRECT_QUEUE_WARN, (msg) => {
    if (!msg) return;
    console.log(`🟡 [WARN] ${msg.content.toString()}`);
    channel.ack(msg);
  });

  // 监听info队列
  channel.consume(DIRECT_QUEUE_INFO, (msg) => {
    if (!msg) return;
    console.log(`🟢 [INFO] ${msg.content.toString()}`);
    channel.ack(msg);
  });
}

Publisher端：发送消息，指定routingKey

// apps/publish/src/routes/direct.route.ts
import { Elysia, t } from "elysia";
import { DIRECT_EXCHANGE_NAME, getRabbitChannel } from "@rabbitmq-demo/shared";

export const directRoutes = new Elysia().post(
  "/direct/publish",
  ({ body }) => {
    const channel = getRabbitChannel();
    channel.publish(
      DIRECT_EXCHANGE_NAME,
      body.level, // routingKey: error / warn / info
      Buffer.from(body.message),
      { persistent: true },
    );
    return {
      status: "ok",
      exchange: DIRECT_EXCHANGE_NAME,
      routingKey: body.level,
      message: body.message,
    };
  },
  {
    body: t.Object({
      level: t.String(),
      message: t.String(),
    }),
  }
);

测试验证

# 发送error消息，只会进入error队列
curl -X POST [http://localhost:3000/direct/publish](http://localhost:3000/direct/publish) -d '{"level":"error","message":"数据库连接失败"}'

# 发送info消息，只会进入info队列
curl -X POST [http://localhost:3000/direct/publish](http://localhost:3000/direct/publish) -d '{"level":"info","message":"服务启动成功"}'

当发送一个不存在的routingKey，消息没有匹配队列，消息直接丢弃。

Direct交换机重点注意事项

匹配规则严格完全相等，大小写敏感，Error 和 error 视为两个不同key。
支持一个队列绑定多个routingKey：

// 同一个队列，同时监听 error 和 warn
await channel.bindQueue(DIRECT_QUEUE_ERROR, DIRECT_EXCHANGE_NAME, "error");
await channel.bindQueue(DIRECT_QUEUE_ERROR, DIRECT_EXCHANGE_NAME, "warn");

幂等：assertExchange、assertQueue 重复调用不会报错，已存在直接跳过，适合多服务启动。
和WorkQueue对比：WorkQueue是一个队列多个消费者；Direct可以做到按key分发到不同队列。

Topic 交换机

Topic交换机是模糊匹配路由，在Direct精准匹配基础上增加通配符支持。routingKey使用.分割成多个单词，支持两种通配符：

*：匹配单个单词
#：匹配0个或多个单词

适合日志、事件分类。例如：sys.error、user.login.info、order.pay.warn。 Direct只能全等匹配；Topic支持模糊匹配，是最灵活的交换机类型。

flowchart LR P["Publisher"] -- "publish routingKey=order.pay.error" --> E["交换机 hmall.topic"] E -- "bindingKey=order.*" --> Q1["队列 order.queue"] E -- "bindingKey=*.error" --> Q2["队列 error.queue"] E -- "bindingKey=#.warn" --> Q3["队列 warn.queue"] Q1 --> C1["消费者：所有订单相关事件"] Q2 --> C2["消费者：全部错误事件"] Q3 --> C3["消费者：所有警告事件"]

routingKey 规则说明

routingKey格式规范：单词用英文点.分隔，单词不能包含*和#

order.*：匹配 order.xxx，只能一层，order.pay.error无法匹配
order.#：匹配 order 开头所有，order、order.pay、order.pay.error 全都匹配
*.error：匹配 xxx.error，一层前缀
#.error：匹配任意层级，结尾是error

限制：#只能单独作为一个段，不能写or#er；*也必须单独一段。

Consumer端 Topic 示例代码

// apps/consumer/src/listeners/topic.listener.ts
import {
  TOPIC_EXCHANGE_NAME,
  TOPIC_QUEUE_ORDER,
  TOPIC_QUEUE_ALL_ERROR,
  type Channel,
} from "@rabbitmq-demo/shared";

export async function startTopicListeners(channel: Channel) {
  await channel.assertExchange(TOPIC_EXCHANGE_NAME, "topic", {
    durable: true,
  });

  await channel.assertQueue(TOPIC_QUEUE_ORDER, { durable: true });
  await channel.assertQueue(TOPIC_QUEUE_ALL_ERROR, { durable: true });

  // 绑定：order.* 匹配order后面单个段
  await channel.bindQueue(TOPIC_QUEUE_ORDER, TOPIC_EXCHANGE_NAME, "order.*");
  // 绑定：*.error 匹配任意一级前缀，后缀error
  await channel.bindQueue(TOPIC_QUEUE_ALL_ERROR, TOPIC_EXCHANGE_NAME, "*.error");

  channel.consume(TOPIC_QUEUE_ORDER, (msg) => {
    if (!msg) return;
    console.log(`📦 [订单事件] ${msg.content.toString()}`);
    channel.ack(msg);
  });

  channel.consume(TOPIC_QUEUE_ALL_ERROR, (msg) => {
    if (!msg) return;
    console.log(`❌ [全局错误] ${msg.content.toString()}`);
    channel.ack(msg);
  });
}

Publisher发送消息

// apps/publish/src/routes/topic.route.ts
import { Elysia, t } from "elysia";
import { TOPIC_EXCHANGE_NAME, getRabbitChannel } from "@rabbitmq-demo/shared";

export const topicRoutes = new Elysia().post(
  "/topic/publish",
  ({ body }) => {
    const channel = getRabbitChannel();
    channel.publish(
      TOPIC_EXCHANGE_NAME,
      body.routingKey,
      Buffer.from(body.message),
      { persistent: true },
    );
    return {
      status: "ok",
      exchange: TOPIC_EXCHANGE_NAME,
      routingKey: body.routingKey,
      message: body.message,
    };
  },
  {
    body: t.Object({
      routingKey: t.String(),
      message: t.String(),
    }),
  }
);

测试用例

# routingKey=order.pay → 匹配 order.* → 进入订单队列
curl -X POST [http://localhost:3000/topic/publish](http://localhost:3000/topic/publish) -d '{"routingKey":"order.pay","message":"订单支付成功"}'

# routingKey=user.error → 匹配 *.error → 进入全局错误队列
curl -X POST [http://localhost:3000/topic/publish](http://localhost:3000/topic/publish) -d '{"routingKey":"user.error","message":"用户登录异常"}'

# routingKey=order.pay.error → order.*无法匹配（两层），只有 *.error 匹配
curl -X POST [http://localhost:3000/topic/publish](http://localhost:3000/topic/publish) -d '{"routingKey":"order.pay.error","message":"订单支付失败"}'

Topic交换机要点

*匹配一个单词，#匹配0或多个单词；
如果bindingKey写成#，等价于Fanout交换机，全部消息都会收到；
如果bindingKey不带通配符，Topic退化成Direct交换机，变成全等匹配；
适合事件驱动系统，业务事件分层（领域.操作.级别）。

四种交换机汇总对比

交换机类型	匹配规则	RoutingKey	典型场景
Default 默认交换机	隐式绑定，sendToQueue语法糖	队列名称	简单点对点任务、WorkQueue
Fanout	广播全部绑定队列	忽略routingKey	发布订阅，多服务同时接收同一条消息，如配置变更通知
Direct	严格全等匹配routingKey	必须完全一致	日志分级、精准路由
Topic	通配符模糊匹配，`.`分段	支持`*` `#`	事件总线，多维度事件分发

消息可靠性（消息不丢失全套机制）

消息丢失的3个风险点：生产者丢失消息、交换机/队列丢失消息、消费者丢失消息

1. 生产者确认（Publisher Confirms）：保证消息成功投递到Broker

默认发布消息是单向发送，生产者发完就认为成功，网络抖动时消息丢了生产者感知不到。开启publisher confirm机制：消息写入Broker磁盘后，Broker给生产者返回ack，生产者收到ack才算投递成功。

失败场景：网络断连、交换机不存在，可以重试投递。

2. 持久化：Exchange、Queue、Message

exchange: {durable:true} 交换机持久化，重启Broker交换机定义不消失
queue: {durable:true} 队列持久化，重启Broker队列定义不消失
发布消息时配置persistent:true，消息持久化到磁盘
⚠️ 三者需要配套；只有队列durable，消息不persistent，消息重启仍然丢失。

3. 消费者手动ACK（手动应答）

noAck:false（默认false），业务代码处理完成之后，手动调用channel.ack(msg)

消费者进程崩溃、程序异常，未ack消息会自动放回队列，重新分发给其他消费者。
noAck:true：消息投递给消费者瞬间直接标记完成，消费者宕机消息直接丢失，生产环境禁止。

注意：不要长时间阻塞不ack，会造成消息堆积。

4. 消息重试与死信队列 DLX（Dead Letter Exchange）

消息变成死信的几种情况：

消息被消费者nack拒绝，并且不重新入队
消息过期（TTL超时）
队列达到最大长度，消息溢出

DLX原理：给普通队列绑定死信交换机，消息成为死信后，自动转发到DLX交换机，路由到死信队列。

用途：失败消息单独归档、延迟重试（RabbitMQ延迟消息经典方案：TTL+DLX）

TTL消息过期

两种设置过期时间：

队列级别：队列内所有消息统一过期时间
消息级别：单条消息单独设置过期时间，优先级高于队列TTL

注意：RabbitMQ的TTL不会主动扫描消息，消息过期后要等到取消息时才判断过期。

消息幂等性

MQ消费天然存在消息重复投递（网络问题，Broker收到消息ack，但ack回传给消费者时网络中断，Broker会重发消息）。

问题：重复消费会造成业务重复执行（例如重复扣款）。

幂等：多次执行同一个消息，业务结果不变。常见实现方案：

唯一消息ID：生产者给每条消息携带业务唯一id，消费前查询数据库/Redis，判断是否已经处理；已处理直接ack跳过。
业务唯一键：利用业务字段天然唯一（订单号）做唯一约束，数据库唯一索引拦截重复。

常见问题与坑点

prefetch是channel级别，不是队列级别；多个队列共用同一个channel，prefetch会互相影响。
assertExchange / assertQueue 是幂等检查；declare类方法只校验，不存在则创建，存在不会覆盖原有配置。
队列没有绑定交换机，消息发送到交换机后直接丢弃。
长连接：Channel基于Connection，一个Connection可以开多个Channel；Channel轻量，推荐每个消费者独立Channel，不要共用。
不要在消费回调里写长时间阻塞IO，会拖慢ack流程，消息堆积。
镜像队列（高可用集群）：单节点宕机，镜像节点接管队列，保证高可用。

项目结构与开发规范

rabbitmq-demo
├── apps
│   ├── publish        # 生产者服务，对外http接口推送消息
│   └── consumer       # 消费者服务，监听队列处理消息
└── shared             # 共享常量：交换机名、队列名、类型定义

规范约定

命名规范：交换机 项目.类型（elysia.fanout）；队列 业务.队列用途；routingKey用小写英文点分隔。
常量抽离：交换机、队列名称统一放到shared包，生产者消费者共用，避免字符串硬编码写错。
生产环境必须：消息持久化 + 手动ack + 生产者confirm，根据业务增加幂等。
异常捕获：消费回调捕获异常，业务失败使用channel.nack(msg, false, true)，可控制是否重回队列。

RabbitMQ 总结

RabbitMQ基于AMQP，核心是交换机+队列的模型：生产者发消息给交换机，交换机按规则路由到队列，消费者监听队列消费。

WorkQueue：点对点任务分发，多消费者竞争消费，重点掌握prefetch+手动ack实现公平调度。
Fanout：广播，一对多，无视routingKey。
Direct：精准字符串匹配路由。
Topic：通配符模糊路由，最通用。

生产落地重点：持久化、手动ack、生产者confirm、消息幂等、死信队列处理失败消息，解决消息丢失、重复消费两大分布式问题

Linzex

RabbitMQ