一、Rab**tMQ架构

Rab**tMQ是一个分布式系统

一、使用rab**tmq时的系统架构图

通过路由键将交换机和队列进行绑定，从而实现消息的发送和接收。

二、rab**tmq基本概念

rab**tmq是AMQP协议的一个开源实现，所以其内部实际上也是AMQP中的基本概念，如下图所示：

1、Message（消息）

消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（传输模式，指出该消息可能需要持久化存储）等。

2、Publisher

消息生产者，也是一个向交换器发布消息的客户端应用程序，就是投递消息的程序。

3、Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。消息交换机，它指定消息按什么规则，路由到哪个队列。

4、Routing Key

路由关键字，exchange根据这个关键字进行消息投递。

5、Binding（绑定）

用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

它的作用就是把exchange和queue按照路由规则绑定起来。

绑定其实就是关联了exchange和queue，或者这么说：queue对exchange的内容感兴趣，exchange要把它的Message deliver到queue。

6、Queue（消息队列）

消息的载体，每个消息都会被投到一个或多个队列，等待消费者连接到这个队列将其取走。它是消息的容器，也是消息的终点。

7、Connection

网络连接，例如一个TCP连接。

8、Channel（信道，通道）

消息通道，在客户端的每个连接里，可建立多个channel。

多路复用连接中的一条独立双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于*作系统来说建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念以达到复用一条TCP连接的目的。

9、Consumer

消息消费者，表示一个从消息队列中取得消息的客户端应用程序，就是接受消息的程序。

10、Virtual Host

虚拟主机，表示一批交换器、消息队列和相关对象。一个broker里可以有多个vhost，用作不同用户的权限分离。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个vhost本质上就是一个mini版的rab**tmq服务器，拥有自己的队列、交换器、绑定和权限机制。

vhost是AMQP概念的基础，必须在连接时指定，rab**tmq默认的vhost是/。

11、Broker

表示消息队列服务器实体。它提供一种传输服务，它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输。

三、AMQP中的消息路由

生产者把消息发布到Exchange上，消息终到达队列并被消费者接收，而Binding决定交换器的消息应该发送到那个队列。如下图所示：

四、Exchange类型

Exchange分发消息时根据类型的不同分发策略有区别，目前共有四种类型：direct、fanout、topic、headers。headers匹配AMQP消息的header而不是路由键，此外headers交换器和direct交换器完全一致，但性能差很多，目前几乎用不到。且看direct、fanout、topic这三种类型。

1、direct类型

消息中的路由键routing key如果和Binding中的**nding key一致，交换器就将消息发到对应的队列中去。路由键与队列名完全匹配，如果一个队列绑定到交换器要求路由键为“dog”，则只转发routing key标记为“dog”的消息，不会转发“dog.puppy”等等。它是完全匹配、单传播的模式。

Driect exchange的路由算法非常简单：通过**ndingkey的完全匹配，可以用下图来说明：

Exchange和两个队列绑定在一起，Q1的**ndingkey是orange，Q2的**nding key是black和green。

当Producer publish key是orange时，exchange会把它放到Q1上，如果是black或green就会到Q2上，其余的Message被丢弃。

2、fanout类型

每个发到fanout类型交换器的消息都会分到所有绑定的队列上去。fanout交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。类似于子网广播，每台子网内的主机都获得了一份**的消息。 fanout类型转发消息是快的。 如下图所示：

3、topic类型

topic交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：#和*，#匹配0个或多个单词，*只能匹配一个单词。

对于Message的routing_key是有限制的，不能是任意的。格式是以点号“.”分割的字符表。比如：”stock.usd.nyse”，“nyse.vmw”，“quick.orange.rab**t”。你可以放任意的key在routing_key中，当然长不能超过255 bytes。对于routing_key，有两个特殊字符#和*，#匹配0个或多个单词，*只能匹配一个单词。如下图所示：

Producer发送消息时需要设置routing_key，routing_key包含三个单词和两个点号，第一个key描述了celerity(灵巧)，第二个是color(色彩)，第三个是物种。

在这里我们创建了两个绑定： Q1的**nding key是”.orange.“； Q2是“..rab**t”和“lazy.#”：Q1感兴趣所有orange颜色的动物；Q2感兴趣所有rab**ts和所有的lazy的。

例如：rounting_key为“quick.orange.rab**t”将会发送到Q1和Q2中。rounting_key为”lazy.orange.rab**t.hujj.ddd”会被投递到Q2中，#匹配0个或多个单词。

五、ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是Rab**tMQ对外提供的API中基本的对象。

1、Connection

Connection是Rab**tmq的socket连接，它封装了socket协议相关部分逻辑。

2、ConnectionFactory

ConnectionFactory是connection的制造工厂。

3、Channel

Channel是我们与rab**tmq打交道的重要的一个接口，大部分的业务*作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

六、任务分发机制

1、Round-ro**n dispathching循环分发

Rabb**tMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的,如果现在load加重,那么只需要创建更多的Consumer来进行任务处理。

2、Message acknowledgment消息确认

为了保证数据不被丢失，Rab**tMQ支持消息确认机制，为了保证数据能被正确处理而不仅仅是被Consumer收到，这就需要在处理完数据之后发送一个确认ack。

在处理完数据之后发送ack，就是告诉Rab**tMQ数据已经被接收并且处理完成，Rab**tMQ可以将消息从队列中移除了。如果Consumer退出了但是没有发送ack，那么Rab**tMQ就会把这个Message发送到下一个Consumer，这样就保证在Consumer异常退出情况下数据也不会丢失。

Rab**tMQ没有用到超时机制，它仅仅通过Consumer的连接中断来确认该Message并没有被正确处理，一个消费者处理消息的时间再长也不会导致该消息被发送给其他消费者，即Rab**tMQ给了Consumer足够长的时间来做数据处理。如果忘记ack，那么当Consumer退出时，Mesage会被重新分发，从而导致队列中的累积的消息越来越多，然后Rab**tMQ会占用越来越多的内存。

3、Message dura**lity消息持久化

如果我们希望即使在rab**tmq服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置成可持久化的（durable），这样就可以保证绝大部分情况下我们的rab**tmq消息不会丢失。但依然解决不了小概率丢失**的发生（例如rab**tmq服务器已经接收到了生产者的消息，但还没来得及持久化该消息时rab**tmq服务器就断电了）。如果也要将这种小概率**管理起来就需要使用到事务了。要持久化队列需要在声明时指定durable=True；这里要注意,队列的名字一定要是Broker中不存在的，不然不能改变此队列的任何属性。队列和交换机有一个创建时候指定的标志durable，durable的唯一含义就是让具有这个标志的队列和交换机会在重启之后重新建立。

消息持久化包括3部分

（1）exchange持久化，在声明时指定durable=> true

channel.ExchangeDeclare(ExchangeName,"direct", durable:true, autoDelete:false, arguments:null);//声明消息队列，且为可持久的

（2）queue持久化，在声明时指定durable=> true

channel.QueueDeclare(QueueName, durable:true, exclusive:false, autoDelete:false, arguments:null);//声明消息队列，且为可持久的

（3）消息持久化，在投递时指定delivery_mode=> 2(1是非持久化)。

channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

如果exchange和queue都是持久化的，那么它们之间的**nding也是持久化的；如果exchange和queue两者之间有一个持久化，一个非持久化,则不允许建立绑定。

注意：一旦创建了队列和交换机，就不能修改其标志了。例如创建了一个non-durable的队列，然后想把它改变成durable的，唯一的办法就是删除这个队列然后重新创建。

4、Fair dispath公平分发

你可能也注意到了，分发机制不是那么优雅，默认状态下，Rab**tMQ将第n个Message分发给第n个Consumer。n是取余后的，它不管Consumer是否还有unacked Message，只是按照这个默认的机制进行分发。那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却毫无休息的机会，那么Rab**t是如何处理这种问题呢?

通过basic.qos方法设置prefetch_count=1，如下设置

channel.basic_qos(prefetch_count=1)

这样Rab**tMQ就会使得每个Consumer在同一个时间点多处理一个Message，换句话说，在接收到该Consumer的ack前，它不会将新的Message分发给它。但是这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。

5、分发到多个Consumer

Direct Exchange：直接匹配，通过Exchange名称+RountingKey来发送与接收消息。

Fanout Exchange：广播订阅，向所有的消费者发布消息，但是只有消费者将队列绑定到该路由器才能收到消息，忽略Routing Key。

Topic Exchange：主题匹配订阅，这里的主题指的是RoutingKey，RoutingKey可以采用通配符，如：*或#，RoutingKey命名采用英文句点来分隔多个词，只有消息将队列绑定到该路由器且指定RoutingKey符合匹配规则时才能收到消息。

Headers Exchange：消息头订阅，消息发布前为消息定义一个或多个键值对的消息头，然后消费者接收消息，同时需要定义类似的键值对请求头（如

x-mactch=all或者x_match=any），只有请求头与消息头匹配，才能接收消息,忽略RoutingKey。

默认的exchange：如果用空字符串去声明一个exchange，那么系统就会使用”amq.direct”这个exchange。我们创建一个queue时，默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去。如下：

channel.BasicPublish("","TaskQueue", properties, bytes);

因为在第一个参数选择了默认的exchange，而我们声明的队列叫TaskQueue，所以默认的，它要新建一个也叫TaskQueue的routingKey，并绑定在默认的exchange上，导致了我们可以在第二个参数routingKey中写TaskQueue，这样它就会找到定义的同名的queue并把消息放进去。

如果有两个接收程序都是用了同一个的queue和相同的routingKey去绑定direct exchange的话，分发的行为是负载均衡的，也就是说第一个是程序1收到，第二个是程序2收到，以此类推。

如果有两个接收程序用了各自的queue，但使用相同的routingKey去绑定direct exchange的话，分发的行为是**的，即每个程序都会收到这个消息的副本。行为相当于fanout类型的exchange。

多个queue绑定同一个key也是可以的，对于下图的例子，Q1和Q2都绑定了black，对于routing key是black的Message，会被deliver到Q1和Q2，其余的Message都会被丢弃。

七、RPC远程过程调用

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到Rab**tMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在Rab**tMQ中也支持RPC。

Rab**tMQ中实现RPC的机制如下图所示：

客户端发送请求（消息）时，在消息的属性（MessageProperties ，在AMQP 协议中定义了14种properties ，这些属性会随着消息一起发送）中设置两个值replyTo （一个Queue 名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue 中）和correlationId （此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行或执行失败）。

服务器端收到消息并处理，服务器端处理完消息后，将生成一条应答消息到replyTo 指定的Queue中，同时带上correlationId 属性，客户端之前已订阅replyTo 指定的Queue ，从中收到服务器的应答消息后，根据其中的correlationId 属性分析哪条请求被执行了，然后根据执行结果进行后续业务处理。

转发：

二、Rab**tMQ*********介绍

各组件解释如下：

AMQP消息的路由中增加了 Exchange和 Binding的角色。生产者把消息发布到 Exchange上，消息终到达队列并被消费者接收，而 Binding决定交换器的消息应该发送到那个队列。

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers。headers匹配 AMQP消息的 header而不是路由键，此外 headers交换器和 direct交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型：

消息中的路由键（routing key）如果和 Binding中的 **nding key一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为"dog"，则只转发 routing key标记为"dog"的消息，不会转发"dog.puppy"，也不会转发"dog.guard"等等。它是完全匹配、单播的模式。

每个发到 fanout类型交换器的消息都会分到所有绑定的队列上去。fanout交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份**的消息。fanout类型转发消息是快的。

topic交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号"#"和符号""。

"#"匹配0个或多个单词，""匹配不多不少一个单词。

Rabb**tMQ的分发机制非常适合扩展，而且它是专门为并发程序设计的，如果现在 load加重，那么只需要创建更多的 Consumer来进行任务处理。

在实际应用中，可能会发生消费者收到 Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给 Rab**tMQ，Rab**tMQ收到消息回执（Message acknowledgment）后才将该消息从 Queue中移除；如果 Rab**tMQ没有收到回执并检测到消费者的 Rab**tMQ连接断开，则 Rab**tMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在 timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的 Rab**tMQ连接断开。这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给 Rab**tMQ，这将会导致严重的 bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑。

另外 pub message是没有 ack的。

如果我们希望即使在 Rab**tMQ服务重启的情况下，也不会丢失消息，我们可以将 Queue与 Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的 Rab**tMQ消息不会丢失。但依然解决不了小概率丢失**的发生（比如 Rab**tMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时 Rab**tMQ服务器就断电了），如果我们需要对这种小概率**也要管理起来，那么我们要用到事务。由于这里仅为 Rab**tMQ的简单介绍，所以这里将不讲解 Rab**tMQ相关的事务。

要持久化队列 queue的持久化需要在声明时指定 durable=True;

这里要注意，队列的名字一定要是 Broker中不存在的，不然不能改变此队列的任何属性.

队列和交换机有一个创建时候指定的标志 durable,durable的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立，它不表示说在队列中的消息会在重启后恢复。

消息持久化包括 3部分

如果 exchange和 queue都是持久化的，那么它们之间的 **nding也是持久化的，如果 exchange和 queue两者之间有一个持久化，一个非持久化，则不允许建立绑定.

注意：一旦创建了队列和交换机，就不能修改其标志了，例如，创建了一个 non-durable的队列，然后想把它改变成 durable的，唯一的办法就是删除这个队列然后重现创建。

你可能也注意到了，分发机制不是那么优雅，默认状态下，Rab**tMQ将第 n个 Message分发给第 n个 Consumer。n是取余后的，它不管 Consumer是否还有 unacked Message，只是按照这个默认的机制进行分发.

那么如果有个 Consumer工作比较重，那么就会导致有的 Consumer基本没事可做，有的 Consumer却毫无休息的机会，那么，Rab**t是如何处理这种问题呢？

Rab**tMQ使用 ProtoBuf序列化消息，它可作为 Rab**tMQ的 Message的数据格式进行传输，由于是结构化的数据，这样就极大的方便了 Consumer的数据高效处理，当然也可以使用 XML，与 XML相比，ProtoBuf有以下优势:

三、Rab**tMQ消费者注意点

    消费者客户端可以通过推模式和拉模式来进行消息消费。

    当rab**tmq队列有多个消费者时，队列收到的消息将以轮询(round-ro**n)的分发方式发送给消费者。每条消息只会发送给订阅列表里的一个消费者。如果现在负载加重，那么只需要创建更多的消费者来消费处理消息即可。

    很多时候轮询的分发机制也不是那么优雅。默认情况下，如果有n个消费者，那么Rab**tMQ会将第m条消息分发给第m%n（取余的方式）个消费者，Rab**tMQ不管消费者是否消费并已经确认（Basic.Ack）了消息。试想一下，如果某些消费者任务繁重，来不及消费那么多的消息，而某些其他消费者由于某些原因（比如业务逻辑简单、机器性能卓越等）很快地处理完了所分配到的消息，进而进程空闲，这样就会造成整体应用吞吐量的下降。

    比如在订阅消费队列之前，消费端程序调用了channel.basicQos(5)，之后订阅了某个队列进行消费。Rab**tMQ会保存一个消费者的列表，每发送一条消息都会为对应的消费者计数，如果达到了所设定的上限，那么 Rab**tMQ就不会向这个消费者再发送任何消息。直到消费者确认了某条消息之后 Rab**tMQ将相应的计数减1，之后消费者可以继续接收消息，直到再次到达计数上限。

    对于channel.basicQos还有要特别注意的一点是它的重载方法

    global为true的时候，是该信道上所有的消费者未确认的消息数上限总和

    为false的时候是针对单个消费者

比如：一个信道设置了

    那么这里每个消费者多只能收到3个未确认的消息，两个消费者能收到的未确认的消息个数之和的上限为5。

如无特殊需要，好只使用global=false设置，这也是默认的设置。

    消息的顺序性是指消费者消费到的消息和发送者发布的消息的顺序是一致的。

    如果生产者发布的消息分别为 msg1 msg2 msg3，那么消费者必然也是按照 msg1 msg2 msg3的顺序进行消费的。

    在以下情况下消息的顺序性会被打破：

    1.如果生产者使用了事务机制，在发送消息之后遇到异常进行了事务回滚，那么需要重新补偿发送这条消息，如果补偿发送是在另一个线程实现的，那么消息在生产者这个源头就出现了错序。

    2.同样，如果启用 publisher confirm时，在发生超时、中断，又或者是收到Rab**tMQ的 Basic.Nack命令时，那么同样需要补偿发送，结果与事务机制一样会错序。或者这种说法有些牵强，我们可以固执地认为消息的顺序性保障是从存入队列之后开始的，而不是在发送的时候。

    3.考虑另一种情形，如果生产者发送的消息设置了不同的超时时间，并且也设置了死信队列，整体上来说相当于一个延迟队列，那么消费者在消费这个延迟队列的时候，消息的顺序必然不会和生产者发送消息的顺序一致。

    4.如果消息设置了优先级，那么消费者消费到的消息也必然不是顺序性的

    5.使用Basic.Nack/.Reject将消息拒绝，比如：

    如果一个队列按照前后顺序分有msg1、msg2、msg3、msg4这4个消息，同时有ConsumerA和 ConsumerB这两个消费者同时订阅了这个队列。队列中的消息轮询分发到各个消费者之中， ConsumerA中的消息为 msg1和msg3，ConsumerB中的消息为msg2、msg4。ConsumerA收到消息 msg1之后并不想处理而调用了Basic.Nack/.Reject将消息拒绝，与此同时将 requeue设置为 true，这样这条消息就可以重新存入队列中。消息 msg1之后被发送到了 ConsumerB中，此时 ConsumerB已经消费了msg2、msg4，之后再消费 msg1，这样消息顺序性也就错*了。或者消息 msg1又重新发往 ConsumerA中，此时 ConsumerA已经消费了msg3，那么再消费 msg1，消息顺序性也无法得到保障。

QueueingConsumer还包含（但不仅限于）以下一些缺陷∶

    一般消息中间件的消息传输保障分为三个层级。

rab**tmq支持少一次和多一次。

少一次：

（1）消息生产者需要开启事务机制或者 publisher confim机制，以确保消息可以可靠地传输到Rab**tMQ中。

（2）消息生产者需要配合使用 mandatory参数或者备份交换器来确保消息能够从交换器路由到队列中，进而能够保存下来而不会被丢弃。

（3）消息和队列都需要进行持久化处理，以确保 Rab**tMQ服务器在遇到异常情况时不会造成消息丢失

（4）消费者在消费消息的同时需要将 autoAck设置为 false，然后通过手动确认的方式去确认已经正确消费的消息，以避免在消费端引起不必要的消息丢失。

多一次：

就无须考虑以上那些方面，生产者随意发送，消费者随意消费，不过这样很难确保消息不会丢失

恰好一次：

恰好一次是Rab**tMQ目前无法保障的。