调度器与进程模型

RabbitMQ 的高并发能力来源于 Erlang VM 的轻量级进程模型与多核调度器。

Erlang 进程模型

进程特性

RabbitMQ 运行在 Erlang VM（BEAM）上，Erlang 进程与操作系统进程不同：

Erlang 进程是用户态轻量级线程，由 Erlang 调度器管理，不依赖 OS 线程。

进程创建与通信

import com.rabbitmq.client.*;

public class ErlangProcessExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            
            // 每个连接在 RabbitMQ 内部对应一个 Erlang 进程
            // 每个 Channel 对应一个独立的 Erlang 进程
            Channel ch1 = conn.createChannel();
            Channel ch2 = conn.createChannel();
            
            // Channel 之间隔离，各自维护独立状态
            ch1.queueDeclare("ch1.queue", true, false, false, null);
            ch2.queueDeclare("ch2.queue", true, false, false, null);
            
            System.out.println("每个 Channel 对应 Broker 内部一个 Erlang 进程");
            
            ch1.close();
            ch2.close();
        }
    }
}

调度器机制

多核调度器

Erlang VM 默认每个 CPU 核心运行一个调度器线程：

CPU 0: [Scheduler 0] → 运行进程 A, B, C
CPU 1: [Scheduler 1] → 运行进程 D, E, F
CPU 2: [Scheduler 2] → 运行进程 G, H, I
...

调度器工作机制：

1. 每个调度器绑定到独立 CPU 核心
2. 调度器从本地运行队列获取进程执行
3. 进程执行固定时间片（约 2000 次函数调用）后让出
4. 进程放回运行队列尾部

任务窃取

当某个调度器队列为空时：

Scheduler 0: [A, B, C]  ← 有任务
Scheduler 1: []          ← 空闲，执行窃取
           ↓
Scheduler 1 从 Scheduler 0 队列尾部窃取进程 C

任务窃取规则：

1. 空闲调度器从最繁忙的调度器队列尾部窃取进程
2. 窃取后双方队列长度趋于平衡
3. 窃取操作加锁，保证线程安全
4. 窃取频率受 +sbwt 参数控制

任务窃取从队列尾部而非头部窃取，避免与调度器本地操作冲突，减少锁竞争。

RabbitMQ 进程架构

核心进程类型

进程间通信

RabbitMQ 内部进程通过 Erlang 消息传递通信：

% 队列进程向消息存储进程请求写入消息
MsgStorePid ! {store_message, Msg, self()}.
% 等待异步响应
receive
    {stored, MsgId} -> ok
end.

Java 客户端侧无需关心内部进程通信，但需理解：

• 每个 Channel 操作通过 Erlang 消息异步发送到 Broker
• Broker 内部进程异步处理请求，返回结果

并发模型

多 Channel 并发

import com.rabbitmq.client.*;
import java.util.concurrent.*;

public class MultiChannelExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setRequestedChannelMax(100);  // 最大 Channel 数
        
        try (Connection conn = factory.newConnection()) {
            ExecutorService executor = Executors.newFixedThreadPool(10);
            
            // 10 个线程各自创建独立 Channel 并发发布消息
            for (int i = 0; i < 10; i++) {
                final int id = i;
                executor.submit(() -> {
                    try {
                        Channel ch = conn.createChannel();
                        ch.queueDeclare("concurrent.queue", true, false, false, null);
                        
                        for (int j = 0; j < 100; j++) {
                            ch.basicPublish("", "concurrent.queue", null,
                                ("Msg-" + id + "-" + j).getBytes());
                        }
                        
                        System.out.println("Thread-" + id + " 完成");
                        ch.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
            
            executor.shutdown();
            executor.awaitTermination(30, TimeUnit.SECONDS);
        }
    }
}

Channel 线程安全

Channel 不是线程安全的。多线程共享同一 Channel 会导致状态混乱。

每个线程应使用独立的 Channel，Connection 是线程安全的，可多线程共享。

调度调优

关键参数

调优建议

1. 调度器数量 = CPU 核心数，避免超线程核心参与调度
2. 连接数建议 ≤ 1000，每连接 Channel 数 ≤ 100
3. 队列数过多时（>10000），考虑增加节点分散负载
4. 高吞吐场景增加 +sbwt 参数，减少任务窃取频率

注意事项

Erlang 进程不是 OS 进程，单个 RabbitMQ 节点可运行数百万进程，无需担心进程数过多。

Channel 非线程安全，多线程并发发布消息时应为每个线程创建独立 Channel。

任务窃取在 CPU 负载不均衡时触发，正常高负载场景窃取操作极少发生。

调度器绑定到 CPU 核心，容器环境需正确设置 +S 参数，否则可能检测到全部宿主机核心。

要点总结

• RabbitMQ 运行在 Erlang VM 上，使用轻量级 Erlang 进程（~300 bytes）
• 每 CPU 核心运行一个调度器，进程时间片约 2000 次函数调用
• 任务窃取机制：空闲调度器从繁忙队列尾部窃取进程，平衡负载
• 每个连接、Channel、队列、交换器对应独立 Erlang 进程
• Channel 非线程安全，多线程应使用独立 Channel
• 调度器数量应等于 CPU 核心数，容器环境需手动设置

文章存放路径：D:\git2\jwdev\articles\RABBITMQ\专家\底层原理与架构\调度器与进程模型.md