为什么要使用消息队列？

总结一下，主要三点原因：解耦、异步、削峰。

1、解耦。比如，用户下单后，订单系统需要通知库存系统，假如库存系统无法访问，则订单减库存将失败，从而导致订单操作失败。订单系统与库存系统耦合，这个时候如果使用消息队列，可以返回给用户成功，先把消息持久化，等库存系统恢复后，就可以正常消费减去库存了。

2、异步。将消息写入消息队列，非必要的业务逻辑以异步的方式运行，不影响主流程业务。

3、削峰。消费端慢慢的按照数据库能处理的并发量，从消息队列中慢慢拉取消息。在生产中，这个短暂的高峰期积压是允许的。比如秒杀活动，一般会因为流量过大，从而导致流量暴增，应用挂掉。这个时候加上消息队列，服务器接收到用户的请求后，首先写入消息队列，如果消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面。

# 使用了消息队列会有什么缺点

• 系统可用性降低。引入消息队列之后，如果消息队列挂了，可能会影响到业务系统的可用性。
• 系统复杂性增加。加入了消息队列，要多考虑很多方面的问题，比如：一致性问题、如何保证消息不被重复消费、如何保证消息可靠性传输等。

# 常见的消息队列对比

总结：

• ActiveMQ 的社区算是比较成熟，但是较目前来说，ActiveMQ 的性能比较差，而且版本迭代很慢，不推荐使用。
• RabbitMQ 在吞吐量方面虽然稍逊于 Kafka 和 RocketMQ ，但是由于它基于 erlang 开发，所以并发能力很强，性能极其好，延时很低，达到微秒级。但是也因为 RabbitMQ 基于 erlang 开发，所以国内很少有公司有实力做 erlang 源码级别的研究和定制。如果业务场景对并发量要求不是太高（十万级、百万级），那这四种消息队列中，RabbitMQ 一定是你的首选。如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。
• RocketMQ 阿里出品，Java 系开源项目，源代码我们可以直接阅读，然后可以定制自己公司的 MQ，并且 RocketMQ 有阿里巴巴的实际业务场景的实战考验。RocketMQ 社区活跃度相对较为一般，不过也还可以，文档相对来说简单一些，然后接口这块不是按照标准 JMS 规范走的有些系统要迁移需要修改大量代码。还有就是阿里出台的技术，你得做好这个技术万一被抛弃，社区黄掉的风险，那如果你们公司有技术实力我觉得用 RocketMQ 挺好的
• Kafka 的特点其实很明显，就是仅仅提供较少的核心功能，但是提供超高的吞吐量，ms 级的延迟，极高的可用性以及可靠性，而且分布式可以任意扩展。同时 kafka 最好是支撑较少的 topic 数量即可，保证其超高吞吐量。kafka 唯一的一点劣势是有可能消息重复消费，那么对数据准确性会造成极其轻微的影响，在大数据领域中以及日志采集中，这点轻微影响可以忽略这个特性天然适合大数据实时计算以及日志收集。

# 如何保证消息队列的高可用？

RabbitMQ：镜像集群模式

RabbitMQ 是基于主从做高可用性的，Rabbitmq有三种模式：单机模式、普通集群模式、镜像集群模式。单机模式一般在生产环境中很少用，普通集群模式只是提高了系统的吞吐量，让集群中多个节点来服务某个 Queue 的读写操作。那么真正实现 RabbitMQ 高可用的是镜像集群模式。

镜像集群模式跟普通集群模式不一样的是，创建的 Queue，无论元数据还是Queue 里的消息都会存在于多个实例上，然后每次你写消息到 Queue 的时候，都会自动和多个实例的 Queue 进行消息同步。这样设计，好处在于：任何一个机器宕机不影响其他机器的使用。坏处在于：1. 性能开销太大：消息同步所有机器，导致网络带宽压力和消耗很重；2. 扩展性差：如果某个 Queue 负载很重，即便加机器，新增的机器也包含了这个 Queue 的所有数据，并没有办法线性扩展你的 Queue。

Kafka：partition 和 replica 机制

Kafka 基本架构是多个 broker 组成，每个 broker 是一个节点。创建一个 topic 可以划分为多个 partition，每个 partition 可以存在于不同的 broker 上，每个 partition 就放一部分数据，这就是天然的分布式消息队列。就是说一个 topic 的数据，是分散放在多个机器上的，每个机器就放一部分数据。

Kafka 0.8 以前，是没有 HA 机制的，任何一个 broker 宕机了，它的 partition 就没法写也没法读了，没有什么高可用性可言。

Kafka 0.8 以后，提供了 HA 机制，就是 replica 副本机制。每个 partition 的数据都会同步到其他机器上，形成自己的多个 replica 副本。然后所有 replica 会选举一个 leader 出来，生产和消费都跟这个 leader 打交道，然后其他 replica 就是 follower。写的时候，leader 会负责把数据同步到所有 follower 上去，读的时候就直接读 leader 上数据即可。Kafka 会均匀的将一个 partition 的所有 replica 分布在不同的机器上，这样才可以提高容错性。

# MQ常用协议

• AMQP协议 AMQP即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。优点：可靠、通用
• MQTT协议 MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器（比如通过Twitter让房屋联网）的通信协议。优点：格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统
• STOMP协议 STOMP（Streaming Text Orientated Message Protocol）是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。优点：命令模式（非topic/queue模式）
• XMPP协议 XMPP（可扩展消息处理现场协议，Extensible Messaging and Presence Protocol）是基于可扩展标记语言（XML）的协议，多用于即时消息（IM）以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输，这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息，即使其操作系统和浏览器不同。优点：通用公开、兼容性强、可扩展、安全性高，但XML编码格式占用带宽大
• 其他基于TCP/IP自定义的协议：有些特殊框架（如：redis、kafka、zeroMq等）根据自身需要未严格遵循MQ规范，而是基于TCP\IP自行封装了一套协议，通过网络socket接口进行传输，实现了MQ的功能。

# MQ的通讯模式

1. 点对点通讯：点对点方式是最为传统和常见的通讯方式，它支持一对一、一对多、多对多、多对一等多种配置方式，支持树状、网状等多种拓扑结构。
2. 多点广播：MQ适用于不同类型的应用。其中重要的，也是正在发展中的是”多点广播”应用，即能够将消息发送到多个目标站点(Destination List)。可以使用一条MQ指令将单一消息发送到多个目标站点，并确保为每一站点可靠地提供信息。MQ不仅提供了多点广播的功能，而且还拥有智能消息分发功能，在将一条消息发送到同一系统上的多个用户时，MQ将消息的一个复制版本和该系统上接收者的名单发送到目标MQ系统。目标MQ系统在本地复制这些消息，并将它们发送到名单上的队列，从而尽可能减少网络的传输量。
3. 发布/订阅(Publish/Subscribe)模式：发布/订阅功能使消息的分发可以突破目的队列地理指向的限制，使消息按照特定的主题甚至内容进行分发，用户或应用程序可以根据主题或内容接收到所需要的消息。发布/订阅功能使得发送者和接收者之间的耦合关系变得更为松散，发送者不必关心接收者的目的地址，而接收者也不必关心消息的发送地址，而只是根据消息的主题进行消息的收发。在MQ家族产品中，MQ Event Broker是专门用于使用发布/订阅技术进行数据通讯的产品，它支持基于队列和直接基于TCP/IP两种方式的发布和订阅。
4. 集群(Cluster)：为了简化点对点通讯模式中的系统配置，MQ提供 Cluster 的解决方案。集群类似于一个 域(Domain) ，集群内部的队列管理器之间通讯时，不需要两两之间建立消息通道，而是采用 Cluster 通道与其它成员通讯，从而大大简化了系统配置。此外，集群中的队列管理器之间能够自动进行负载均衡，当某一队列管理器出现故障时，其它队列管理器可以接管它的工作，从而大大提高系统的高可靠性

# 如何保证消息的顺序性？

查看更多

1.什么是 ActiveMQ?

activeMQ 是一种开源的，实现了 JMS1.1 规范的，面向消息(MOM)的中间件，为应用程序提供高效的、 可扩展的、稳定的和安全的企业级消息通信

2. ActiveMQ 服务器宕机怎么办？  

这得从 ActiveMQ 的储存机制说起。在通常的情况下，非持久化消息是存储在内存中的，持久化消息是存 储在文件中的，它们的最大限制在配置文件的<systemUsage>节点中配置。但是，在非持久化消息堆积  到一定程度，内存告急的时候，ActiveMQ 会将内存中的非持久化消息写入临时文件中，以腾出内存。虽 然都保存到了文件里，但它和持久化消息的区别是，重启后持久化消息会从文件中恢复，非持久化的临时 文件会直接删除。

那如果文件增大到达了配置中的最大限制的时候会发生什么？我做了以下实验：

设置 2G 左右的持久化文件限制，大量生产持久化消息直到文件达到最大限制，此时生产者阻塞，但消费  者可正常连接并消费消息，等消息消费掉一部分，文件删除又腾出空间之后，生产者又可继续发送消息， 服务自动恢复正常。

设置 2G 左右的临时文件限制，大量生产非持久化消息并写入临时文件，在达到最大限制时，生产者阻塞， 消费者可正常连接但不能消费消息，或者原本慢速消费的消费者，消费突然停止。整个系统可连接，但是  无法提供服务，就这样挂了。

具体原因不详，解决方案：尽量不要用非持久化消息，非要用的话，将临时文件限制尽可能的调大。

3. 丢消息怎么办？

这得从 java 的 java.net.SocketException 异常说起。简单点说就是当网络发送方发送一堆数据，然后调   用 close 关闭连接之后。这些发送的数据都在接收者的缓存里，接收者如果调用 read 方法仍旧能从缓存中 读取这些数据，尽管对方已经关闭了连接。但是当接收者尝试发送数据时，由于此时连接已关闭，所以会  发生异常，这个很好理解。不过需要注意的是，当发生 SocketException 后，原本缓存区中数据也作废了， 此时接收者再次调用 read 方法去读取缓存中的数据，就会报 Software caused connection abort: recv  failed 错误。

通过抓包得知，ActiveMQ 会每隔 10 秒发送一个心跳包，这个心跳包是服务器发送给客户端的，用来判   断客户端死没死。如果你看过上面第一条，就会知道非持久化消息堆积到一定程度会写到文件里，这个写 的过程会阻塞所有动作，而且会持续 20 到 30 秒，并且随着内存的增大而增大。当客户端发完消息调用    connection.close()时，会期待服务器对于关闭连接的回答，如果超过 15 秒没回答就直接调用 socket 层  的 close 关闭 tcp 连接了。这时客户端发出的消息其实还在服务器的缓存里等待处理，不过由于服务器心  跳包的设置，导致发生了 java.net.SocketException 异常，把缓存里的数据作废了，没处理的消息全部丢 失。

解决方案：用持久化消息，或者非持久化消息及时处理不要堆积，或者启动事务，启动事务后，commit() 方法会负责任的等待服务器的返回，也就不会关闭连接导致消息丢失了。

4. 持久化消息非常慢。   

默认的情况下，非持久化的消息是异步发送的，持久化的消息是同步发送的，遇到慢一点的硬盘，发送消 息的速度是无法忍受的。但是在开启事务的情况下，消息都是异步发送的，效率会有 2 个数量级的提升。 所以在发送持久化消息时，请务必开启事务模式。其实发送非持久化消息时也建议开启事务，因为根本不 会影响性能。

5. 消息的不均匀消费。   

有时在发送一些消息之后，开启 2 个消费者去处理消息。会发现一个消费者处理了所有的消息，另一个消 费者根本没收到消息。原因在于 ActiveMQ 的 prefetch 机制。当消费者去获取消息时，不会一条一条去  获取，而是一次性获取一批，默认是 1000 条。这些预获取的消息，在还没确认消费之前，在管理控制台  还是可以看见这些消息的，但是不会再分配给其他消费者，此时这些消息的状态应该算作“已分配未消费”， 如果消息最后被消费，则会在服务器端被删除，如果消费者崩溃，则这些消息会被重新分配给新的消费者。  但是如果消费者既不消费确认，又不崩溃，那这些消息就永远躺在消费者的缓存区里无法处理。更通常的 情况是，消费这些消息非常耗时，你开了 10 个消费者去处理，结果发现只有一台机器吭哧吭哧处理，另   外 9 台啥事不干。

解决方案：将 prefetch 设为 1，每次处理 1 条消息，处理完再去取，这样也慢不了多少。

6. 死信队列。   

如果你想在消息处理失败后，不被服务器删除，还能被其他消费者处理或重试，可以关闭

AUTO_ACKNOWLEDGE，将 ack 交由程序自己处理。那如果使用了 AUTO_ACKNOWLEDGE，消息是什 么时候被确认的，还有没有阻止消息确认的方法？有！

消费消息有 2 种方法，一种是调用 consumer.receive()方法，该方法将阻塞直到获得并返回一条消息。这 种情况下，消息返回给方法调用者之后就自动被确认了。另一种方法是采用 listener 回调函数，在有消息 到达时，会调用 listener 接口的 onMessage 方法。在这种情况下，在 onMessage 方法执行完毕后，消 息才会被确认，此时只要在方法中抛出异常，该消息就不会被确认。那么问题来了，如果一条消息不能被 处理，会被退回服务器重新分配，如果只有一个消费者，该消息又会重新被获取，重新抛异常。就算有多 个消费者，往往在一个服务器上不能处理的消息，在另外的服务器上依然不能被处理。难道就这么退回–  获取–报错死循环了吗？

在重试 6 次后，ActiveMQ 认为这条消息是“有毒”的，将会把消息丢到死信队列里。如果你的消息不见 了，去 ActiveMQ.DLQ 里找找，说不定就躺在那里。

7. ActiveMQ 中的消息重发时间间隔和重发次数吗？

查看更多

问题一：RabbitMQ  中的 broker   是指什么？ cluster   又是指什么？  

答： broker  是指一个或多个 erlang node   的逻辑分组，且 node   上运行着 RabbitMQ  应用 程序。 cluster  是在 broker   的基础之上，增加了 node  之间共享元数据的约束。

问题二：什么是元数据？元数据分为哪些类型？包括哪些内容？与 cluster    相关的元数据 有哪些？元数据是如何保存的？元数据在 cluster   中是如何分布的？

答：在非  cluster    模式下，元数据主要分为  Queue     元数据（queue     名字和属性等）、 Exchange   元数据（exchange    名字、类型和属性等）、Binding    元数据（存放路由关系的查 找表）、Vhost       元数据（vhost       范围内针对前三者的名字空间约束和安全属性设置）。在 cluster  模式下，还包括 cluster   中 node   位置信息和 node   关系信息。元数据按照 erlang node    的类型确定是仅保存于  RAM    中，还是同时保存在  RAM    和 disk    上。元数据在 cluster  中是全 node  分布的。

下图所示为 queue   的元数据在单 node  和 cluster  两种模式下的分布图。

问题三： RAM node  和 disk node   的区别？ 

答： RAM node  仅将 fabric（即 queue 、exchange   和 binding 等 RabbitMQ 基础构件）相 关元数据保存到内存中，但 disk  node   会在内存和磁盘中均进行存储。 RAM  node   上唯一 会存储到磁盘上的元数据是 cluster   中使用的 disk node   的地址。要求在 RabbitMQ cluster 中至少存在一个 disk node  。

问题四： RabbitMQ  上的一个 queue   中存放的 message   是否有数量限制？  

答：可以认为是无限制，因为限制取决于机器的内存，但是消息过多会导致处理效率的下 降。

问题五： RabbitMQ   概念里的 channel 、exchange   和 queue   这些东东是逻辑概念，还是对 应着进程实体？这些东东分别起什么作用？  

答： queue   具有自己的 erlang  进程； exchange   内部实现为保存 binding   关系的查找表； channel     是实际进行路由工作的实体，即负责按照  routing_key      将  message     投递给 queue    。由  AMQP     协议描述可知， channel     是真实  TCP    连接之上的虚拟连接，所有 AMQP  命令都是通过 channel   发送的，且每一个 channel   有唯一的 ID 。一个 channel   只 能被单独一个操作系统线程使用，故投递到特定 channel   上的 message   是有顺序的。但 一个操作系统线程上允许使用多个 channel   。channel   号为 0   的 channel   用于处理所有 对于当前 connection  全局有效的帧，而 1-65535   号 channel   用于处理和特定 channel   相 关的帧。 AMQP  协议给出的 channel   复用模型如下

其中每一个 channel  运行在一个独立的线程上，多线程共享同一个 socket。

问题六： vhost  是什么？起什么作用？  

答： vhost    可以理解为虚拟  broker     ，即  mini-RabbitMQ     server 。其内部均含有独立的 queue 、exchange    和  binding     等，但最最重要的是，其拥有独立的权限系统，可以做到 vhost   范围的用户控制。当然，从 RabbitMQ    的全局角度， vhost   可以作为不同权限隔离 的手段（一个典型的例子就是不同的应用可以跑在不同的 vhost  中）。

【cluster  相关】

问题七：在单 node   系统和多 node   构成的 cluster   系统中声明 queue 、exchange   ，以及 进行 binding  会有什么不同？

答：当你在单 node   上声明 queue   时，只要该 node   上相关元数据进行了变更，你就会 得到 Queue.Declare-ok    回应；而在 cluster   上声明 queue    ，则要求  cluster   上的全部 node  都要进行元数据成功更新，才会得到 Queue.Declare-ok   回应。另外，若 node   类型 为 RAM node  则变更的数据仅保存在内存中，若类型为 disk node   则还要变更保存在磁盘 上的数据。

问题八：客户端连接到 cluster   中的任意 node   上是否都能正常工作？  

答：是的。客户端感觉不到有何不同。

问题九：若 cluster   中拥有某个 queue   的 owner  node   失效了，且该 queue  被声明具有 durable  属性，是否能够成功从其他 node  上重新声明该 queue   ？

答：不能，在这种情况下，将得到 404  NOT_FOUND   错误。只能等 queue    所属的 node 恢复后才能使用该 queue   。但若该 queue   本身不具有 durable   属性，则可在其他 node 上重新声明。

问题十： cluster   中 node   的失效会对 consumer   产生什么影响？若是在 cluster   中创建了 mirrored queue  ，这时 node  失效会对 consumer  产生什么影响？  

答：若是 consumer   所连接的那个 node   失效（无论该 node   是否为 consumer   所订阅 queue   的 owner  node），则 consumer   会在发现 TCP   连接断开时，按标准行为执行重连 逻辑，并根据“Assume   Nothing”原则重建相应的  fabric     即可。若是失效的  node     为 consumer   订阅 queue    的 owner  node，则 consumer    只能通过 Consumer  Cancellation Notification   机制来检测与该 queue   订阅关系的终止，否则会出现傻等却没有任何消息来 到的问题。

问题十一：能够在地理上分开的不同数据中心使用 RabbitMQ cluster  么？ 

答：不能。第一，你无法控制所创建的 queue  实际分布在 cluster  里的哪个 node   上（一 般使用  HAProxy   +   cluster     模型时都是这样），这可能会导致各种跨地域访问时的常见问 题；第二， Erlang    的 OTP   通信框架对延迟的容忍度有限，这可能会触发各种超时，导致

业务疲于处理；第三，在广域网上的连接失效问题将导致经典的“脑裂”问题，而 RabbitMQ   目前无法处理（该问题主要是说 Mnesia）。

【综合问题】 

问题十二：为什么 heavy RPC   的使用场景下不建议采用 disk node   ？

答： heavy RPC  是指在业务逻辑中高频调用 RabbitMQ  提供的 RPC  机制，导致不断创建、 销毁 reply queue   ，进而造成 disk node   的性能问题（因为会针对元数据不断写盘）。所以 在使用 RPC  机制时需要考虑自身的业务场景。

问题十三：向不存在的 exchange   发 publish   消息会发生什么？向不存在的 queue   执行consume  动作会发生什么？

答：都会收到 Channel.Close  信令告之不存在（内含原因 404 NOT_FOUND）。

问题十四： routing_key  和 binding_key   的最大长度是多少？

答： 255  字节。

问题十五： RabbitMQ  允许发送的 message   最大可达多大？

查看更多