MQ 的 Erlang 使用思路：从并发模型到消息队列实践-拓冰建站

1. 引言：为什么选择 Erlang 构建 MQ？

消息队列（Message Queue, MQ）是现代分布式系统的核心组件，负责解耦服务、异步通信、削峰填谷和保证最终一致性。在众多编程语言和框架中，Erlang/OTP因其独特的并发模型、容错机制和热代码升级能力，成为构建高可靠、高并发消息中间件的理想选择。

本文将系统梳理使用 Erlang 设计和实现 MQ 的核心思路，涵盖并发模型、进程设计、消息传递、持久化策略以及集群管理等方面，为开发者提供从理论到实践的完整指引。

2. Erlang 的核心优势与 MQ 的契合点

2.1 轻量级进程（Process）与 Actor 模型

Erlang 的进程是语言级轻量级实体，创建开销极小（约 300 字节），调度由虚拟机（BEAM）负责。这种设计使得每个连接、每个队列甚至每个消息都可以由一个独立的进程来管理，天然支持海量并发。

隔离性：进程间内存完全隔离，一个进程崩溃不会影响其他进程，符合 MQ 需要高可用的要求。
无锁并发：基于消息传递（Share Nothing），避免锁竞争，简化并发编程。

2.2 软实时与低延迟

BEAM 虚拟机的软实时调度器优先保证响应性，通过减少垃圾回收（GC）停顿（每个小进程独立 GC），非常适合需要稳定低延迟的消息投递场景。

2.3 内置分布式与容错（OTP）

OTP（Open Telecom Platform）提供了一套成熟的监督树（Supervision Tree）模式：

监督者（Supervisor）：监控工作进程（Worker），失败后按策略重启。
应用（Application）：管理启动、停止和依赖。
GenServer、GenStatem：标准化服务器和状态机行为，简化队列、交换机等核心组件的实现。

这为 MQ 的可靠性提供了开箱即用的基础设施。

2.4 热代码升级（Hot Code Swapping）

Erlang 支持在不停止服务的情况下更新代码版本，这对于需要 7x24 小时运行的 MQ 服务至关重要，可以实现无缝升级和 bug 修复。

3. 使用 Erlang 设计 MQ 的核心思路

3.1 架构分层

一个典型的 Erlang MQ 可分为以下层次：

协议层：解析 AMQP、MQTT、STOMP 等协议，每个连接一个进程（gen_server）。
会话层：管理连接状态、认证、通道（Channel）。
核心路由层：实现交换机（Exchange）、队列（Queue）、绑定（Binding）和消息路由逻辑。
存储层：消息持久化（可选），可集成 Mnesia、ETS/DETS、外部数据库（如 PostgreSQL、RocksDB）。
集群管理层：基于 Erlang 分布式节点（net_kernel、global）或第三方方案（如riak_core）实现节点发现、数据分片和故障转移。

3.2 进程模型设计

为每个核心实体分配专属进程：

队列进程：管理消息的入队、出队、持久化和订阅者列表。可使用gen_server维护状态（消息列表、消费者 PID 等）。
交换机进程：根据类型（direct、topic、fanout）和绑定规则，将消息路由到目标队列进程。
连接进程：处理客户端 TCP/SSL 连接，解析协议帧，并创建通道进程。
通道进程：在连接内提供轻量级并发，执行发布、消费、确认等命令。

进程间通过异步消息（!）通信，例如：协议层收到发布命令 → 发送消息到交换机进程 → 交换机根据绑定转发到队列进程 → 队列进程存储并通知消费者进程。

3.3 消息存储与持久化策略

根据消息的持久化（Persistent）标志和可靠性要求，可选择不同存储方案：

内存存储（ETS）：用于非持久化消息或临时队列，速度快，进程崩溃后丢失。
磁盘存储（DETS/Mnesia）：Erlang 内置数据库，适合中小规模持久化，支持事务和分布式表。
外部存储：对于海量消息，可集成 RocksDB（通过erlang-rocksdb）或 Kafka 作为后端存储引擎。
混合存储：热数据在内存（ETS），冷数据异步刷盘，通过gen_statem管理状态转换。

3.4 集群与高可用

Erlang 分布式节点可通过net_kernel:connect/1组成集群。MQ 集群设计需考虑：

队列镜像：主队列进程在多个节点上启动从镜像，通过pg2或global进程组管理成员，主节点故障时自动切换。
数据分片：将队列哈希到不同节点，避免单点瓶颈，可使用riak_core实现一致性哈希环。
网络分区处理：通过net_kernel:monitor_nodes/1监听节点状态，采用自动愈合或手动干预策略。

4. 实践示例：一个简单的内存队列实现

以下是一个用 Erlang/OTP 实现的简易内存队列gen_server，展示核心思路：

-module(simple_queue). -behaviour(gen_server). %% API -export([start_link/0, publish/2, consume/1, ack/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -record(state, { messages = [] :: list(), % 消息列表 [{Id, Msg}] consumers = [] :: list(pid()), % 消费者进程列表 next_id = 1 :: integer() % 下一条消息ID }). %% 启动队列进程 start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %% 发布消息 publish(Msg, Opts) -> gen_server:call(?MODULE, {publish, Msg, Opts}). %% 消费消息（阻塞直到有消息） consume(ConsumerPid) -> gen_server:call(?MODULE, {consume, ConsumerPid}). %% 确认消息 ack(ConsumerPid, MsgId) -> gen_server:cast(?MODULE, {ack, ConsumerPid, MsgId}). %% 初始化 init([]) -> {ok, #state{}}. %% 处理发布请求 handle_call({publish, Msg, _Opts}, _From, State = #state{messages = Msgs, next_id = Id}) -> NewMsg = {Id, Msg}, NewState = State#state{ messages = Msgs ++ [NewMsg], next_id = Id + 1 }, %% 通知所有消费者 lists:foreach(fun(Pid) -> Pid ! {new_message, NewMsg} end, State#state.consumers), {reply, {ok, Id}, NewState}; %% 处理消费请求 handle_call({consume, ConsumerPid}, _From, State = #state{messages = Msgs, consumers = Consumers}) -> case Msgs of [] -> %% 无消息，将消费者加入等待列表 NewState = State#state{consumers = [ConsumerPid | Consumers]}, {reply, empty, NewState}; [Msg | Rest] -> NewState = State#state{messages = Rest}, {reply, {ok, Msg}, NewState} end; handle_call(_Request, _From, State) -> {reply, {error, unknown_call}, State}. %% 处理确认（异步） handle_cast({ack, _ConsumerPid, _MsgId}, State) -> %% 这里可以记录确认日志或清理消息 {noreply, State}; handle_cast(_Msg, State) -> {noreply, State}. %% 处理消费者退出等消息 handle_info({'DOWN', _Ref, process, Pid, _Reason}, State = #state{consumers = Consumers}) -> NewConsumers = lists:delete(Pid, Consumers), {noreply, State#state{consumers = NewConsumers}}; handle_info(_Info, State) -> {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}.

这个示例展示了：