RabbitMQ源码分析 – 实体初始化

作者：你好赵伟 | 2024-06-14 06:37:55

踩

初始化rabbitmqtemplate的实体类

（注：分析代码基于RabbitMQ 2.8.2）

Connection

在上篇文章中提到，客户端连上rabbit后，需要向rabbit发送AMQP协议头，rabbit在收到协议头后，开始在0号channel 上跟客户端进行交互（AMQP中一个连接可以多路复用，1~65535为可用的channel编号，0号channel，也就是frame中channel的索引为0，被认为是全局于整个连接）。

根据AMQP协议，经过 connection.start -> connection.start_ok -> connection.secure -> connection.secure_ok -> connection.tune -> connection.tune_ok（这时rabbit会建立一个心跳进程）-> connection.open -> connection.open_ok后，客户端与rabbit之间就认为已经建立了连接（相关代码参见[$RABBIT_SRC/src/rabbit_reader.erl --> handle_method0/2]）。

（粗体指令由rabbit服务器发出）

Channel

当客户端发来的frame中，channel的索引不为0时，rabbit认为这些数据从属于某个channel。如果该channel进程不存在，则会创建一个channel进程（rabbit_channel，具体参见[$RABBIT_SRC/src/rabbit_reader.erl --> create_channel/2），并由此进程负责该channel上的所有数据（[$RABBIT_SRC/src/rabbit_reader.erl --> process_channel_frame/3）。

根据AMQP协议，经过channel.open -> channel.open_ok（[$RABBIT_SRC/src/rabbit_channel.erl --> handle_method/3）后，客户端就可以开始在该channel上发送数据了。

Exchange

当rabbit收到来自客户端的exchange.declare指令时，rabbit会根据客户端的参数创建一个exchange。首先rabbit会向mnesia的表rabbit_exchange写入一条记录，包含客户端请求的exchange类型信息（默认4种类型：direct，topic，fanout，head）及相关参数，如果exchange是需要持久化的（durable），则还需要向rabbit_durable_exchange表中写入相同信息。然后，rabbit会通过rabbit_event发送exchange_created的事件（统计作用）。

Queue

queue在创建时，需要确定要创建queue的类型（rabbit里称为backing_queue）：一般情况下，queue只在当前结点（客户端所连接的结点）创建，对应backing_queue为rabbit_variable_queue；当有 HA策略时，queue需要在集群中的多个结点上创建（这时候，有master结点和slave结点之分，master结点未必是当前结点），master结点创建队列对应backing_queue为rabbit_mirror_queue_master，slave结点对应backing_queue为rabbit_mirror_queue_slave（master，slave实际最终也会创建一个rabbit_variable_queue）。

创建一个队列首先会创建一个rabbit_amqqueue_process进程。然后同exchange类似，都需要先在mnesia表里写入queue的基本信息（rabbit_queue，rabbit_durable_queue）。然后初始化对应的backing_queue，最后发送queue_create事件。

我们来看一下backing_queue为rabbit_variable_queue时，初始化需要做什么：1）初始化queue的索引（rabbit_queue_index:init/2）或者从以前的队列恢复索引（rabbit_queue_index:recover/5，durable队列）；2）创建message store或者恢复message store（rabbit_msg_store:client_init/4，恢复仅对于durable队列）。

rabbit_mirror_queue_master初始化：1）创建一个rabbit_mirror_queue_coordinator及相应GM（Guaranteed Multicast）；2）获取该队列相关的镜像节点，并调用rabbit_mirror_queue_misc:add_mirror/2启动镜像队列进程；3）在当前结点（或者master结点）初始化一个rabbit_variable_queue队列；4）通过GM向所有镜像广播3中初始化队列的长度。

rabbit_mirror_queue_misc:add_mirror/2启动镜像队列进程时，启动的是一个rabbit_mirror_queue_slave进程，相比rabbit_variable_queue，它只是多了个初始化GM的工作。最后也会初始化一个rabbit_variable_queue队列。

Binding

用于将queue绑定到一个exchange。主要涉及到几个数据表写入（见下表，true或者false代表相应对象是不是durable），无其它复杂逻辑，写入完成后会发送binding_created事件。

exchage queue table

true true rabbit_durable_route

rabbit_semi_durable_route

rabbit_route

rabbit_reverse_route

false true rabbit_semi_durable_route

rabbit_route

rabbit_reverse_route

true/false false rabbit_route

rabbit_reverse_route

topic类型的exchange还需要将binding信息写入以下数据表：rabbit_topic_trie_edge，rabbit_topic_trie_binding，rabbit_topic_trie_node（基于 trie数据结构，用于route key的匹配）。

（本文只是个流程上的梳理，文中的一些概念及作用后续会再详细分析）

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/716849