【烈日炎炎战后端】消息队列（1.0万字）

消息队列

1. 消息队列的优点有哪些？
2. 消息队列的缺点有哪些？
3. 如何保证消息的有序性？
4. 如何保证消息的可靠性传输？
5. RabbitMQ如何实现消息确认机制？
6. 如何保证消息队列的高可用？
7. RabbitMQ/ActiveMQ/RocketMQ/Kafka对比
8. RabbitMQ/ActiveMQ/RocketMQ/Kafka如何选择

1. 消息队列的优点有哪些？

消息队列的主要作用是：解耦、异步、削峰.

解耦：**消费者和生产者互不影响，降低他们之间的耦合度.**A 系统通过接口调用发送数据到 B、C、D 三个系统.那如果现在 E 系统也要这个数据呢？那如果 C 系统现在不需要了呢？现在 A 系统又要发送第二种数据了呢？这样的话 A 系统的维护成本就非常的高，而且 A 系统要时时刻刻考虑B、C、D、E 四个系统如果出现故障该怎么办？A 系统是重发还是先把消息保存起来呢？使用消息队列就可以解决这个问题.A 系统只负责生产数据，不需要考虑消息被哪个系统来消费.
异步：**生产者生产完消息直接放入消息队列中，不需要等待结果，继续处理其他.**A 系统需要发送个请求给 B 系统处理，由于 B 系统需要查询数据库花费时间较长，以至于 A 系统要等待 B 系统处理完毕后再发送下个请求，造成 A 系统资源浪费.使用消息队列后，A 系统生产完消息后直接丢进消息队列，不用等待 B 系统的结果，直接继续去干自己的事情了.
削峰：**在请求数量太大导致系统无法承受的时候可以先放进消息队列再根据能力接受请求.**A 系统调用 B 系统处理数据，每天 0 点到 12 点，A 系统风平浪静，每秒并发请求数量就 100 个.结果每次一到 12 点 ~ 13 点，每秒并发请求数量突然会暴增到 1 万条.但是 B 系统最大的处理能力就只能是每秒钟处理 1000 个请求，这样系统很容易就会崩掉.这种情况可以引入消息队列，把请求数据先存入消息队列中，消费系统再根据自己的消费能力拉取消费.

2. 消息队列的缺点有哪些？

消息队列可能会引发：降低系统可用性、系统复杂度提高、一致性问题.

降低系统的可用性：系统引入的外部依赖越多，越容易挂掉；
系统复杂度提高：使用 MQ 后可能需要保证消息没有被重复消费、处理消息丢失的情况、保证消息传递的顺序性等等问题；
一致性问题：A 系统处理完了直接返回成功了，但问题是：要是 B、C、D 三个系统那里，B 和 D 两个系统写库成功了，结果 C 系统写库失败了，就造成数据不一致了.

3. 如何保证消息的有序性？

Kafka：拆分多个 Queue，每个 Queue一个 Consumer（消费者），就是多一些 Queue 而已，确实是麻烦点；或者就一个 Queue 但是对应一个 Consumer，然后这个 Consumer 内部用内存队列做排队，然后分发给底层不同的 Worker 来处理.
RabbitMQ：一个 Topic，一个 Partition，一个 Consumer，内部单线程消费，单线程吞吐量太低，一般不会用这个；写 N 个内存 Queue，具有相同 key 的数据都到同一个内存 Queue；然后对于 N 个线程，每个线程分别消费一个内存 Queue 即可，这样就能保证顺序性.

应用场景方面
RabbitMQ：用于实时的，对可靠性要求较高的消息传递上.
kafka：用于处于活跃的流式数据，大数据量的数据处理上.
架构模型方面
producer，broker，consumer
RabbitMQ：以broker为中心，有消息的确认机制
kafka：以consumer为中心，无消息的确认机制
.吞吐量方面
RabbitMQ：支持消息的可靠的传递，支持事务，不支持批量操作，基于存储的可靠性的要求存储可以采用内存或硬盘，吞吐量小.
kafka：内部采用消息的批量处理，数据的存储和获取是本地磁盘顺序批量操作，消息处理的效率高，吞吐量高.
.集群负载均衡方面
RabbitMQ：本身不支持负载均衡，需要loadbalancer的支持
kafka：采用zookeeper对集群中的broker，consumer进行管理，可以注册topic到zookeeper上，通过zookeeper的协调机制，producer保存对应的topic的broker信息，可以随机或者轮询发送到broker上，producer可以基于语义指定分片，消息发送到broker的某个分片上.

4. 如何保证消息的可靠性传输？

一、如果是生产者弄丢了数据

生产者将数据发送到RabbitMQ的时候，可能数据就在半路给搞丢了，因为网络啥的问题，都有可能。

此时可以选择用RabbitMQ提供的事务功能，就是生产者发送数据之前开启RabbitMQ事务（channel.txSelect），然后发送消息，如果消息没有成功被RabbitMQ接收到，那么生产者会收到异常报错，此时就可以回滚事务（channel.txRollback），然后重试发送消息；如果收到了消息，那么可以提交事务（channel.txCommit）。这种方式会降低吞吐量，影响性能。

还有一种方式是开启confirm模式，在生产者那里设置开启confirm模式之后，你每次写的消息都会分配一个唯一的id，然后如果写入了RabbitMQ中，RabbitMQ会给你回传一个ack消息，告诉你说这个消息ok了。如果RabbitMQ没能处理这个消息，会回调你一个nack接口，告诉你这个消息接收失败，你可以重试。而且你可以结合这个机制自己在内存里维护每个消息id的状态，如果超过一定时间还没接收到这个消息的回调，那么你可以重发。

**对比：**事务机制和cnofirm机制最大的不同在于，事务机制是同步的，你提交一个事务之后会阻塞在那儿，但是confirm机制是异步的，你发送个消息之后就可以发送下一个消息，然后那个消息RabbitMQ接收了之后会异步回调你一个接口通知你这个消息接收到了。所以一般在生产者这块避免数据丢失，都是用confirm机制的。

二、如果是RabbitMQ弄丢了数据

就是RabbitMQ自己弄丢了数据，这个你必须开启RabbitMQ的持久化，消息写入之后会持久化到磁盘，哪怕是RabbitMQ自己挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢。除非极其罕见的是，RabbitMQ还没持久化，自己就挂了，可能导致少量数据会丢失的，但是这个概率较小。

设置持久化有两个步骤，第一个是创建queue的时候将其设置为持久化的，这样就可以保证RabbitMQ持久化queue的元数据，但是不会持久化queue里的数据；第二个是发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化的，此时RabbitMQ就会将消息持久化到磁盘上去。

必须要同时设置这两个持久化才行，RabbitMQ哪怕是挂了，再次重启，也会从磁盘上重启恢复queue，恢复这个queue里的数据。而且持久化可以跟生产者那边的confirm机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者ack了，所以哪怕是在持久化到磁盘之前，RabbitMQ挂了，数据丢了，生产者收不到ack，你也是可以自己重发的。

给RabbitMQ开启了持久化机制后，也还有一种可能，就是这个消息写到了RabbitMQ中，但是还没来得及持久化到磁盘上，结果不巧，此时RabbitMQ挂了，就会导致内存里的一点点数据会丢失。

三、消费端弄丢了数据

消费的时候，刚消费到，还没处理，结果进程挂了，比如重启了，RabbitMQ认为你都消费了，这数据就丢了。这个时候得用RabbitMQ提供的ack机制，简单来说，就是你关闭RabbitMQ自动ack，可以通过一个api来调用就行，然后每次你自己代码里确保处理完的时候，再程序里ack一把。

这样的话，如果你还没处理完，不就没有ack？那RabbitMQ就认为你还没处理完，这个时候RabbitMQ会把这个消费分配给别的consumer去处理，消息是不会丢的。

消费端弄丢了数据

你消费到了这个消息，然后消费者那边自动提交了offset，让kafka以为你已经消费完了这个消息，其实你刚准备处理这个消息，你还没处理，你自己就挂了，此时这条消息就丢了。

那么只要关闭自动提交offset，在处理完之后自己手动提交offset，就可以保证数据不会丢。

但是此时确实还是会重复消费，比如你刚处理完，还没提交offset，结果自己挂了，此时肯定会重复消费一次，自己保证幂等性就好了。

kafka弄丢了数据

这块比较常见的一个场景，就是kafka某个broker宕机，然后重新选举partiton的leader时。要是此时其他的follower刚好还有些数据没有同步，结果此时leader挂了，然后选举某个follower成leader之后，他不就少了一些数据？

生产环境也遇到过，我们也是，之前kafka的leader机器宕机了，将follower切换为leader之后，就会发现说这个数据就丢了所以此时一般是要求起码设置如下4个参数：

在topic设置replication.factor参数：这个值必须大于1，要求每个partition必须有至少2个副本

在kafka服务端设置min.insync.replicas参数：这个值必须大于1，这个是要求一个leader至少感知到有至少一个follower还跟自己保持联系，没掉队，这样才能确保leader挂了还有一个follower

在producer端设置acks=all：这个是要求每条数据，必须是写入所有replica之后，才能认为是写成功了

在producer端设置retries=MAX（无限次重试的意思）：这个是要求一旦写入失败，就无限重试，卡在这里了

我们生产环境就是按照上述要求配置的，这样配置之后，至少在kafka broker端就可以保证在leader所在broker发生故障，进行leader切换时，数据不会丢失

生产者会不会弄丢数据

如果按照上述的思路设置了ack=all，一定不会丢，要求是，你的leader接收到消息，所有的follower都同步到了消息之后，才认为本次写成功了。如果没满足这个条件，生产者会自动不断的重试，重试无限次。

5. RabbitMQ如何实现消息确认机制？

https://www.cnblogs.com/DBGzxx/p/10091070.html

当消息的投送方把消息投递出去，却不知道消息是否投递成功了.如果消息投送方不管的话，势必对系统的造成可靠性的影响.
可是如果要保证系统的可靠性，消息投靠方，如何知道消息是否投放成功了呢？
这个就需要消息的确认机制，我们来看下rabbitMQ的消息去人机制是如何做的.

消息的确认分两部分：rabbitMQ确认生产者投递的消息和消费者确认 rabbitMQ服务器的消息
1. 首先说RabbitMQ对生产者的确认，总共分为两种模式分别为同步模式与异步模式.
  
  (1) 同步模式分为单条消息确认与批量确认.
  
  ① 单条消息确认： channel.waitForConfirms() 普通发送方确认模式；消息到达交换器，就会返回true.
  
  ② 批量消息确认： channel.waitForConfirmsOrDie()批量确认模式；使用同步方式等所有的消息发送之后才会执行后面代码，只要有一个消息未到达交换器就会抛出IOException异常.
  
  （2) 异步模式为生产者异步监听消息确认.
异步监听消息确认：channel.addConfirmListener()异步监听发送方确认模式.
```
  　　 3. 其次说下消费者对RabbitMQ 消息确认.总共分为两种方式 分别为 手动确认 和 自动确认.
```
消费者收到的每一条消息都必须进行确认.消息确认后，RabbitMQ才会从队列删除这条消息，RabbitMQ不会为未确认的消息设置超时时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消的消费者连接是否已经断开.

这么设计的原因是RabbitMQ允许消费者消费一条消息的时间可以很久很久.

(1)　自动确认：

消费者在声明队列时，可以指定autoAck参数，当autoAck=true时，一旦消费者接收到了消息，就视为自动确认了消息.如果消费者在处理消息的过程中，出了错，就没有什么办法重新处理这条消息，所以我们很多时候，

需要在消息处理成功后，再确认消息，这就需要手动确认.

(2) 手动确认：

①当autoAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息.否则，RabbitMQ会在队列中消息被消费后立即删除它.

②采用消息确认机制后，只要令autoAck=false，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为RabbitMQ会一直持有消息直到消费者显式调用basicAck为止.

③当autoAck=false时，对于RabbitMQ服务器端而言，队列中的消息分成了两部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者ack信号的消息.如果服务器端一直没有收到消费

者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者）.

④通过运行程序，启动两个消费者A、B，都可以收到消息，但是其中有一个消费者A不会对消息进行确认，当把这个消费者A关闭后，消费者B又会收到本来发送给消费者A的消息.所以我们一般使用手动确认的方法是，将消息的处理放在try/catch语句块中，成功处理了，就给RabbitMQ一个确认应答，如果处理异常了，就在catch中，进行消息的拒绝

附录：

RabbitMQ

Kafka

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

根据不同的 MQ 或者你用过的 MQ 进行回答：

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 分布在不同的机器上，这样才可以提高容错性.

7. RabbitMQ/ActiveMQ/RocketMQ/Kafka对比

https://www.cnblogs.com/Terry-Wu/p/7644279.html

8. RabbitMQ/ActiveMQ/RocketMQ/Kafka如何选择

https://www.cnblogs.com/chjxbt/p/11394185.html

一、选择消息队列产品的基本标准

在消息队列的技术选型上，并不存在说哪个消息队列就是“最好的”.常用的几个消息队列，每个产品都有自己的优势和劣势，需要根据现有系统的情况，选择最适合的那款产品.

技术产品的及格标准：

必须是开源产品：如果遇到Bug至少有机会通过修改源代码迅速修复或规避，解决燃眉之急.
必须是近年来比较流行并且有一定社区活跃度的产品：流行的好处是，只要使用的场景不太冷门，遇到的Bug都可以找到解决办法.
流行的产品与周边生态系统会有一个比较好的集成和兼容：比如kafka和Flink就有比较好的兼容性，Flink内置了kafka的Data Sourse，使得你不用自己开发一个Flink的Data Source.

消息队列产品的及格标准：

消息的可靠传递：确保不丢消息.
Cluster：支持集群，确保不会因为某个节点宕机导致服务不可用，当然也不能丢消息.
性能：具备足够好的性能，能满足绝大多数场景的性能要求.

二、可供选择的消息队列产品

1、RabbitMQ

介绍：

使用Erlang语言编写，最早是为电信行业系统之间的可靠通讯性设计的，也是少数几个支持AMQP协议的消息队列.
轻量级、迅速，开箱即用，非常容易部署和使用.
有一个特色的功能是支持非常灵活的路由配置.它在生产者和队列之间增加了一个Exchange模块，可以理解为交换机，根据配置的路由规则将生产者发出的消息分发到不同的队列中.
客户端支持的编程语言是所有消息队列中最多的.

劣势：

对消息堆积的支持并不好.在它的设计理念中，消息队列是一个管道，大量的消息积压不是正常的情况，应当尽量避免.当大量消息积压时，会导致性能急剧下降.
RabbitMQ的性能是介绍的几个消息队列中最差的，它大概每秒可以处理几万到几十万条消息，这个性能也足够支撑绝大多数场景.不过，如果你的应用对消息队列的性能要求非常高，那就不要选择RabbitMQ.
小众的编程语言Erlang.如果想基于RabbitMQ做一些扩展和二次开发，建议你慎重考虑可持续维护的问题.

RabbitMQ 官方文档：https://www.rabbitmq.com/documentation.html

2、RocketMQ

介绍：

RocketMQ是阿里巴巴2012年开源的产品，后来捐赠给Apache软件基金会.2017年正式成为Apache的顶级项目.
阿里内部也是使用RocketMQ作为支撑其业务的消息队列，经历多次“双十一”考验，它的性能、稳定性和可靠性都是值得信赖的.
有非常活跃的中文社区，大多数问题都可以找到中文答案.使用Java语言开发，它的贡献者大多数为中国人，源代码相对比较易懂或易进行扩展和二次开发.
RocketMQ对在线业务的响应时延做了很多优化，大多数情况下可以做到毫秒级的响应，如果你的应用场景很在意响应时延，那应该选择使用RocketMQ.
RocketMQ的性能比RabbitMQ要高一个数量级，每秒大概能处理几十万条消息.

劣势：

作为国产消息队列，相比国外比较流行的同类产品，与周边生态系统的集成和兼容程度要略逊一筹.

RocketMQ 官方文档：https://rocketmq.apache.org/docs/quick-start/

RocketMQ 中国开发者中心：http://rocketmq.cloud/zh-cn/

3、Kafka

介绍：

最早是由LinkedIn开发，目前也是Apache的顶级项目.最初的设计目的是用于处理海量的日志.
与周边生态系统的兼容性是最好的，尤其在大数据和流计算领域，几乎所有的相关开源软件系统都会优化支持kafka.
使用Scala和Java语言开发，设计上大量使用了批量和异步的思想，这种设计使得Kafka能做到超高的性能，尤其是异步收发性能，是三者中最好.
大概每秒可处理几十万条消息

劣势：

同步收发消息的响应时延比较高，当你的业务场景中，每秒消息数量没那么多时，kafka的时延反而会比较高.所以不太适合在线业务场景.

Kafka 官方文档：http://kafka.apache.org/documentation/

三、第二梯队的消息队列

这些产品之所有没那么流行，或多或少都有着比较明显的短板，不推荐使用，只是作简单介绍.

1、ActiveMQ

最老牌的开源消息队列，目前已进入老年期，社区不活跃.
功能或性能方面都与现代的消息队列存在明显的差距，它存在的意义仅限于兼容还在用的爷爷辈系统.

2、ZeroMQ

严格来说ZeroMQ并不能称之为一个消息队列，而是一个基于消息队列的多线程网络库.
如果你需要将消息队列的功能集成到你的系统进程中，可以考虑使用ZeroMQ.

3、Pulsar

一个新兴的开源消息队列产品，最早由Yahoo开发，目前处于成长期，流行度和成熟度相对没有那么高.
采用存储和计算分离的设计，有可能会引领未来消息队列的一个发展方向，建议持续关注这个项目.

四、总结：

选择的建议：

如果消息队列并不是将要构建系统的主角之一，对消息队列功能和性能都没有很高的要求，只需一个开箱即用易维护的产品，建议使用RabbitMQ.(有良好的运维界面，仅仅只是使用消息队列功能，用于异步和业务模块解耦，对性能要求不是很高.rabbitMQ能满足现阶段需求)

如果系统使用消息队列的场景是处理在线业务(在线业务指的是那种服务于web页面或者APP的服务，这种服务都需要很低的延迟，否则APP就会很卡，体验不好)，比如在交易系统中用消息队列传递订单，那RocketMQ的低延迟和金融级的稳定性是你需要的.
如果需要处理海量的消息，想收集日志、监控信息或是前端埋点这类数据，或应用场景大量使用了大数据、流计算(做事后的统计分析)相关的开源产品，那kafka是最适合的.