flowchart TB
    Client[客户端连接<br/>AMQP/MQTT/STOMP/Stream] --> |协议帧| Listener[TCP Listener]
    Listener --> |创建连接| Reader[Connection Reader<br/>rabbit_reader]
    
    Reader --> |解析协议| Channel[Channel 进程<br/>rabbit_channel]
    Reader -.-> |AMQP 1.0| AmqpSession[AMQP 1.0 Session<br/>rabbit_amqp_session]
    
    Channel --> |声明/绑定| Exchange[Exchange<br/>rabbit_exchange]
    Channel --> |发布消息| Router[Router<br/>rabbit_router]
    Channel --> |消费消息| Queue[Queue 进程<br/>各类 queue_type]
    
    Router --> Exchange
    Exchange --> |路由规则| Binding[Binding<br/>rabbit_binding]
    Binding --> Queue
    
    Exchange --> |类型实现| ExDirect[Direct Exchange]
    Exchange --> |类型实现| ExTopic[Topic Exchange]
    Exchange --> |类型实现| ExFanout[Fanout Exchange]
    Exchange --> |类型实现| ExHeaders[Headers Exchange]
    
    Queue --> |经典队列| ClassicQ[rabbit_classic_queue<br/>rabbit_variable_queue]
    Queue --> |仲裁队列| QuorumQ[rabbit_quorum_queue<br/>rabbit_fifo Ra 机器]
    Queue --> |流队列| StreamQ[rabbit_stream_queue<br/>osiris 日志]
    
    ClassicQ --> MsgStore[Message Store<br/>rabbit_msg_store]
    ClassicQ --> QIndex[Queue Index<br/>rabbit_classic_queue_index_v2]
    QuorumQ --> Ra[Ra Consensus<br/>分布式日志]
    StreamQ --> Osiris[Osiris Log<br/>持久化日志]
    
    Channel --> |元数据操作| DB[(Database<br/>Mnesia/Khepri)]
    Exchange --> DB
    Queue --> DB
    
    subgraph 核心监督树
        RabbitSup[rabbit_sup] --> |启动| VHostSupSup[VHost Sup Sup<br/>rabbit_vhost_sup_sup]
        RabbitSup --> |启动| Networking[Networking<br/>rabbit_networking]
        RabbitSup --> |启动| NodeMonitor[Node Monitor<br/>rabbit_node_monitor]
        
        VHostSupSup --> |每个 vhost| VHostSup[VHost Sup<br/>rabbit_vhost_sup]
        VHostSup --> |每个队列| QueueSup[Queue Sup<br/>rabbit_amqqueue_sup]
    end
    
    subgraph 插件系统
        Management[Management API<br/>rabbitmq_management]
        MQTT[MQTT Adapter<br/>rabbitmq_mqtt]
        STOMP[STOMP Adapter<br/>rabbitmq_stomp]
        StreamPlugin[Stream Plugin<br/>rabbitmq_stream]
        Federation[Federation<br/>rabbitmq_federation]
        Shovel[Shovel<br/>rabbitmq_shovel]
        
        Management -.-> Channel
        MQTT -.-> Channel
        STOMP -.-> Channel
        StreamPlugin -.-> Queue
    end

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Reader as rabbit_reader
    participant Channel as rabbit_channel
    participant Exchange as rabbit_exchange
    participant Router as rabbit_router
    participant Queue as Queue 进程
    participant QBackend as Queue Backend<br/>(variable_queue/ra)
    participant Consumer as Consumer 进程<br/>(Channel)
    
    %% 发布阶段
    Note over Client,Queue: === 消息发布阶段 ===
    Client->>+Reader: basic.publish 帧
    Reader->>Reader: 解析帧
    Reader->>+Channel: 转发方法调用
    Channel->>Channel: 权限检查
    Channel->>+Exchange: route(Message, Opts)
    Exchange->>Exchange: 根据 type 调用路由函数
    Exchange->>-Channel: 返回目标队列列表
    
    Channel->>+Router: deliver_to_queues(Queues, Msg)
    loop 每个目标队列
        Router->>+Queue: deliver(Msg)
        Queue->>+QBackend: publish(Msg)
        QBackend->>QBackend: 写入存储<br/>(内存/磁盘/Raft)
        QBackend-->>-Queue: ok
        Queue-->>-Router: ok
    end
    Router-->>-Channel: delivery_result
    
    Channel->>Channel: 记录 confirms（如需要）
    Channel-->>-Reader: basic.ack (publisher confirm)
    Reader-->>-Client: 确认帧
    
    %% 消费阶段
    Note over Client,Consumer: === 消息消费阶段 ===
    Queue->>Queue: 检查待分发消息
    Queue->>+Consumer: push message
    Consumer->>Consumer: 格式化消息
    Consumer->>Reader: basic.deliver 帧
    Reader->>Client: 推送消息
    
    Client->>+Reader: basic.ack 帧
    Reader->>+Consumer: 确认消息
    Consumer->>+Queue: settle(ack, MsgId)
    Queue->>+QBackend: ack(MsgId)
    QBackend->>QBackend: 删除/标记消息
    QBackend-->>-Queue: ok
    Queue-->>-Consumer: ok
    Consumer-->>-Reader: ok
    Reader-->>-Client: ok

# 方式 1：使用默认配置启动
rabbitmq-server

# 方式 2：前台启动（开发调试）
rabbitmq-server -detached

# 方式 3：Erlang shell 启动（高级）
erl -pa deps/*/ebin -s rabbit

%% 连接到节点
{ok, Conn} = amqp_connection:start(#amqp_params_direct{}).

%% 打开通道
{ok, Channel} = amqp_connection:open_channel(Conn).

%% 声明队列
QDeclare = #'queue.declare'{queue = <<"test_queue">>},
#'queue.declare_ok'{} = amqp_channel:call(Channel, QDeclare).

%% 发布消息
Publish = #'basic.publish'{exchange = <<>>, routing_key = <<"test_queue">>},
Msg = #amqp_msg{payload = <<"Hello RabbitMQ">>},
ok = amqp_channel:cast(Channel, Publish, Msg).

%% 消费消息
Subscribe = #'basic.consume'{queue = <<"test_queue">>},
#'basic.consume_ok'{} = amqp_channel:subscribe(Channel, Subscribe, self()).

%% 接收消息
receive
    {#'basic.deliver'{}, #amqp_msg{payload = Payload}} ->
        io:format("Received: ~p~n", [Payload])
end.

%% 关闭连接
amqp_connection:close(Conn).

-module(my_exchange).
-behaviour(rabbit_exchange_type).

-export([description/0, serialise_events/0, route/2, route/3,
         validate/1, validate_binding/2, create/2, delete/2,
         policy_changed/2, add_binding/3, remove_bindings/3,
         assert_args_equivalence/2, info/1, info/2]).

%% 描述
description() ->
    [{description, <<"Custom exchange type">>}].

serialise_events() -> false.

%% 核心路由逻辑
route(#exchange{name = XName}, Msg, _Opts) ->
    %% 自定义路由算法
    RKey = mc:routing_key(Msg),
    %% 示例：哈希路由到特定队列
    HashValue = erlang:phash2(RKey, 10),
    TargetQueue = rabbit_misc:r(XName, queue, integer_to_binary(HashValue)),
    [TargetQueue].

%% 其他回调（省略实现）
validate(_X) -> ok.
validate_binding(_X, _B) -> ok.
create(_Serial, _X) -> ok.
delete(_Serial, _X) -> ok.
policy_changed(_X1, _X2) -> ok.
add_binding(_Serial, _X, _B) -> ok.
remove_bindings(_Serial, _X, _Bs) -> ok.
assert_args_equivalence(X, Args) ->
    rabbit_exchange:assert_args_equivalence(X, Args).
info(_X) -> [].
info(_X, _) -> [].

-rabbit_boot_step({my_exchange_type,
                   [{description, "custom exchange type"},
                    {mfa, {rabbit_registry, register,
                           [exchange, <<"x-custom">>, my_exchange]}},
                    {requires, rabbit_registry},
                    {enables, kernel_ready}]}).

%% 声明自定义类型的 exchange
ExDeclare = #'exchange.declare'{
    exchange = <<"my_custom_exchange">>,
    type = <<"x-custom">>
},
#'exchange.declare_ok'{} = amqp_channel:call(Channel, ExDeclare).

-module(my_auth_backend).
-behaviour(rabbit_authn_backend).
-behaviour(rabbit_authz_backend).

-export([user_login_authentication/2, user_login_authorization/2,
         check_vhost_access/3, check_resource_access/4, check_topic_access/4,
         state_can_expire/0]).

%% 认证
user_login_authentication(Username, AuthProps) ->
    %% 从外部系统验证用户（例如 HTTP API）
    case external_api:verify_user(Username, AuthProps) of
        {ok, UserData} ->
            {ok, #auth_user{username = Username,
                           tags = proplists:get_value(tags, UserData, [])}};
        {error, _} ->
            {refused, "Authentication failed", []}
    end.

%% 授权
user_login_authorization(Username, _AuthProps) ->
    {ok, _Impl} = application:get_env(my_plugin, authz_impl),
    {ok, {Username, _Impl}}.

check_vhost_access(#auth_user{username = Username}, VHost, _Sock) ->
    %% 检查虚拟主机访问权限
    external_api:check_vhost(Username, VHost).

%% 其他授权回调（省略）
check_resource_access(_User, _Resource, _Permission, _Context) ->
    true.

check_topic_access(_User, _Resource, _Permission, _Context) ->
    true.

state_can_expire() -> false.

%% rabbitmq.config
[
  {rabbit, [
    {auth_backends, [my_auth_backend]}
  ]}
].

%% rabbitmq.conf
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_consul
cluster_formation.consul.host = 127.0.0.1
cluster_formation.consul.port = 8500
cluster_formation.consul.svc = rabbitmq
cluster_formation.node_cleanup.only_log_warning = true

%% rabbitmq.conf
vm_memory_high_watermark.relative = 0.6
disk_free_limit.absolute = 10GB
total_memory_available_override_value = 16GB

# 使用 rabbitmqctl 设置内存告警
rabbitmqctl set_vm_memory_high_watermark 0.5

# 检查告警状态
rabbitmqctl alarm_list

# 运行时调整日志级别
rabbitmqctl set_log_level debug

# 恢复默认
rabbitmqctl set_log_level info