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>