核心模块源码结构

RabbitMQ 服务端使用 Erlang 编写，采用 OTP 框架构建。理解源码分层结构是深入源码分析的前提。

源码获取

# 克隆官方仓库
git clone https://github.com/rabbitmq/rabbitmq-server.git

# 切换到目标版本
cd rabbitmq-server
git checkout v3.13.0

顶层目录结构

rabbitmq-server/
├── apps/                    # 核心应用模块
│   ├── rabbit/              # RabbitMQ 核心实现
│   ├── amqp_client/         # AMQP 客户端库（Erlang 版）
│   ├── rabbitmq_management/ # 管理插件
│   └── rabbitmq_common/     # 公共工具库
├── deps/                    # 依赖库
│   ├── amqp10_client/       # AMQP 1.0 客户端
│   ├── cowboy/              # HTTP 服务器
│   └── jsx/                 # JSON 处理
├── plugins/                 # 内置插件
│   ├── rabbitmq_prometheus/
│   ├── rabbitmq_tracing/
│   └── rabbitmq_shovel/
└── test/                    # 测试代码

核心模块 apps/rabbit/

apps/rabbit/
├── src/
│   ├── rabbit_app.erl           # 应用入口
│   ├── rabbit.erl               # 核心公共 API
│   │
│   ├── network/                 # 网络层
│   │   ├── rabbit_reader.erl    # TCP 连接处理
│   │   ├── rabbit_writer.erl    # 数据发送
│   │   └── rabbit_connection.erl # 连接管理
│   │
│   ├── amqp_0_9_1/              # AMQP 0-9-1 协议
│   │   ├── rabbit_framing.erl   # 帧编码/解码
│   │   ├── rabbit_channel.erl   # Channel 核心逻辑
│   │   └── rabbit_command_impl.erl # 命令实现
│   │
│   ├── routing/                 # 路由层
│   │   ├── rabbit_exchange.erl  # Exchange 管理
│   │   ├── rabbit_amqqueue.erl  # Queue 管理
│   │   └── rabbit_binding.erl   # Binding 管理
│   │
│   ├── backing_queue/           # 队列存储
│   │   ├── rabbit_backing_queue.erl    # 存储接口
│   │   ├── rabbit_msg_store.erl        # 消息持久化存储
│   │   └── rabbit_variable_queue.erl   # 可变队列实现
│   │
│   ├── queue_master/            # 队列主控
│   │   └── rabbit_queue_master.erl     # 队列生命周期管理
│   │
│   └── virtual_host/            # 虚拟主机
│       └── rabbit_vhost.erl            # vhost 隔离与资源管理

模块依赖关系

                    ┌─────────────┐
                    │  rabbit_app  │  应用入口
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ network  │ │ routing  │ │  vhost   │
        │  (连接)   │ │ (路由)    │ │ (隔离)    │
        └────┬─────┘ └────┬─────┘ └──────────┘
             │             │
             ▼             ▼
        ┌──────────┐ ┌──────────┐
        │ channel  │ │  queue   │
        │ (通道)    │ │ (队列)    │
        └──────────┘ └────┬─────┘
                          │
              ┌───────────┼───────────┐
              ▼           ▼           ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ backing  │ │ msg_store│ │ mirror   │
        │ _queue   │ │ (持久化)  │ │ (镜像)    │
        └──────────┘ └──────────┘ └──────────┘

关键模块解析

1. 网络层 rabbit_reader.erl

负责 TCP 连接接收、帧解析、协议分发。

% 简化示例：读取 AMQP 帧
recv_frame(Socket) ->
    % 读取帧头（7 bytes）
    {ok, Header} = gen_tcp:recv(Socket, 7),
    % 解析帧类型和大小
    {FrameType, Channel, PayloadSize} = parse_header(Header),
    % 读取负载
    {ok, Payload} = gen_tcp:recv(Socket, PayloadSize),
    {FrameType, Channel, Payload}.

2. 通道层 rabbit_channel.erl

处理 Channel 级别命令：basic_publish、basic_consume、basic_ack 等。

% 处理 basic_publish
handle_method({'basic.publish', _, Exchange, RoutingKey}, Content, State) ->
    % 查找 Exchange
    {ok, Ex} = rabbit_exchange:lookup(State#ch.vhost, Exchange),
    % 路由到匹配的 Queue
    Queues = rabbit_exchange:route(Ex, RoutingKey),
    % 投递消息
    lists:foreach(fun(Q) -> deliver_message(Q, Content) end, Queues),
    {noreply, State}.

3. 路由层 rabbit_exchange.erl

Exchange 类型实现：direct、fanout、topic、headers。

% direct exchange 路由
route(#exchange{type = <<"direct">>, bindings = Bindings}, RoutingKey) ->
    % 精确匹配 routing_key
    [Queue || {BindingKey, Queue} <- Bindings, BindingKey =:= RoutingKey];

% fanout exchange 路由
route(#exchange{type = <<"fanout">>, bindings = Bindings}, _RoutingKey) ->
    % 忽略 routing_key，投递到所有绑定队列
    [Queue || {_BindingKey, Queue} <- Bindings].

4. 存储层 rabbit_msg_store.erl

消息持久化到磁盘，支持索引和恢复。

% 消息写入磁盘
store_message(MsgId, Content) ->
    % 计算存储路径
    Path = store_path(MsgId),
    % 写入文件
    file:write_file(Path, Content, [sync]),
    % 更新索引
    update_index(MsgId, Path).

Erlang 进程模型

RabbitMQ 中每个连接、通道、队列都对应独立的 Erlang 进程：

每个 Erlang 进程轻量（约 300 字节），可支撑百万级并发连接。

源码阅读建议

1. 从 rabbit_app.erl 入口开始，理解应用启动流程
2. 跟踪一条消息的完整路径：rabbit_reader → rabbit_channel → rabbit_exchange → rabbit_amqqueue → rabbit_msg_store
3. 重点关注状态机实现：Channel、Connection、Queue 都是典型的 OTP 状态机
4. 使用 recon 工具进行运行时分析：追踪进程消息传递、内存分配

注意事项

1. RabbitMQ 源码使用 Erlang 编写，需具备 OTP 框架基础才能有效阅读
2. 源码中大量使用 gen_server、gen_statem 等 OTP 行为模式
3. 模块间通过 Erlang 消息传递通信，非共享内存架构
4. 持久化存储层 rabbit_msg_store 使用 Mnesia + 文件混合存储
5. 镜像队列逻辑在 rabbit_mirror_queue_master.erl 中实现，涉及分布式同步

要点总结

• RabbitMQ 源码位于 apps/rabbit/src/ 目录，按网络层、通道层、路由层、存储层分层
• 核心模块依赖：rabbit_app → network → channel → routing → queue → backing_queue
• 每个连接、通道、队列对应独立 Erlang 进程，采用 OTP 监督树管理
• 源码阅读建议从应用入口跟踪一条消息的完整路由路径
• 理解 gen_server 状态机模式和 Erlang 消息传递机制是阅读源码的基础