RabbitMQ架构
<p>RabbitMQ是一个高可用的消息中间件,支持多种协议和集群扩展。并且支持消息持久化和镜像队列,适用于对消息可靠性较高的场合,基本模型如下。</p> <p><img src="https://simg.open-open.com/show/3931949bb4852036ab995b325a182592.png"></p> <p>其客户端使用方式</p> <pre> <code class="language-java">from kombu import Connection, Exchange, Queue media_exchange = Exchange('media', 'direct', durable=True) video_queue = Queue('video', exchange=media_exchange, routing_key='video') def process_media(body, message): print body message.ack() # connections with Connection('amqp://guest:guest@localhost//') as conn: # produce producer = conn.Producer(serializer='json') producer.publish({'name': '/tmp/lolcat1.avi', 'size': 1301013}, exchange=media_exchange, routing_key='video', declare=[video_queue]) # the declare above, makes sure the video queue is declared # so that the messages can be delivered. # It's a best practice in Kombu to have both publishers and # consumers declare the queue. You can also declare the # queue manually using: # video_queue(conn).declare() # consume with conn.Consumer(video_queue, callbacks=[process_media]) as consumer: # Process messages and handle events on all channels while True: conn.drain_events() </code></pre> <p>示例中的发布端和消费端是同一方,而实际中的使用方式一般有多种场景,topic模式、fanout模式、direct模式和RPC模式。</p> <p><img src="https://simg.open-open.com/show/c2cab17b6ee560943403c1d998d7a9d7.png"> <img src="https://simg.open-open.com/show/c813069a23d72a7a3f519d896270a62a.png"></p> <ol> <li>topic模式,按照设置的路由信息(routing key)将消息路由到一个或者多个消费端,而消息只能由一个消费者消费一次。一个消费者可以设置多个路由信息,可以同时获取多个消费者发送的消息;</li> <li>fanout模式,与topic模式唯一的区别是同一消息会发送到订阅(binding)的多个消费者;</li> <li>direct模式,一对一模式,实际中比较少用;</li> <li>RPC模式,结合topic和direct模式,发送消息的同时指定要接受的消息。</li> </ol> <h2>RabbitMQ监控树</h2> <p>为了高可靠,Erlang中实际的工作进程(Erlang进程,并不是系统进程)都有一个监控进程,监控进程负责(一个或多个)工作进程的创建、销毁和重启。监控进程和工作进程的关系如图。</p> <p><img src="https://simg.open-open.com/show/a856d4d7116d9fff97ed4f5644c18bd5.gif"></p> <ol> <li>方块图是监控进程;</li> <li>圆圈是工作进程;</li> <li>方块中的”1“(one_for_one)和”a“(one_for_all)代表不同的监控策略</li> </ol> <p>one_for_one 监控策略,一个工作进程崩溃,则只重启崩溃的工作进程。</p> <p><img src="https://simg.open-open.com/show/63550e4d3caa83db90fd9d99308b0bd0.gif"></p> <p>one_for_all监控策略,一个工作进程崩溃,则销毁并重启所有工作进程</p> <p><img src="https://simg.open-open.com/show/4f9166bd46cbf384d3619869c45f6f5b.gif"> 在RabbitMQ中还有一种 <strong>simple_one_for_one监控策略</strong> ,与 <strong>one_for_one监控策略</strong> 相同,只不过重启工作进程时的启动参数是固定的。RabbitMQ网络框架也遵循该原则。</p> <p><img src="https://simg.open-open.com/show/9205f06d8ed482ea685b6b75f8198252.png"></p> <h2>RabbitMQ消息架构</h2> <p>当client端链接服务器时,RabbitMQ会启动一系列监控和工作进程来处理网络连接。</p> <p><img src="https://simg.open-open.com/show/a2763af3034ab97f9ce717eddecf16c9.png"> 为了降低TCP链接数量,多个消费者共享同一个链接Connection,但是每个消费者独享一个管道channel,用consumer_tag标识。consumer_tag在Connection唯一,从1开始累加,当重连接时需要匹配该tag。每个消费者对应独立的一套rabbit_channel_sup_sup->rabbit_channel_sup->rabbit_channel|rabbit_writer|rabbit_limiter系列进程。</p> <h2>RabbitMQ网络框架时序图</h2> <p><img src="https://simg.open-open.com/show/735933d45bdd9eebdf57e85d8ecc655d.png"> client建立链接后,RabbitMQ通过tcp_acceptor进程处理accept成功后返回的clientfd。</p> <p><img src="https://simg.open-open.com/show/c819cf50f1447e883cdc8944475d5e7a.png"> rabbit_reader从TCP链接中读取数据,然后根据协议回调函数处理客户端的各种请求。</p> <h2>RabbitMQ消息处理流程</h2> <p><img src="https://simg.open-open.com/show/f8eddd13ef623384c0f2754761ed1dbf.png"> RabbitMQ先验证权限;然后检查Exchange是否存在,不存在则创建;检查消息是否合法以及是否需要confirm等;根据路由信息选择消费队列;检查消费队列是否存在,有则将消息发送给消息队列;检查消费者是否存在,存在则将消息发送给消费者client端。</p> <p><img src="https://simg.open-open.com/show/b57c4db61d20b17821a88d408e70d86d.png"> RabbitMQ会根据不同的消息的不同类型做不同的处理:</p> <ol> <li>不持久化消息,如果没有消费者则直接丢掉,不会入消费队列;如果有,则先入消息队列,按照入队顺序依次发送给消费者。</li> <li>持久化消息,将消息持久化成功后才给发送端发ack,然后再发送给消费者。</li> </ol> <p>(完)</p> <p> </p> <p>来自:https://fanchao01.github.io/blog/2018/02/09/rabbitmq-arch/</p> <p> </p>
本文由用户 mssj0912 自行上传分享,仅供网友学习交流。所有权归原作者,若您的权利被侵害,请联系管理员。
转载本站原创文章,请注明出处,并保留原始链接、图片水印。
本站是一个以用户分享为主的开源技术平台,欢迎各类分享!