主流消息队列RabbitMq,RocketMq,Kafka

着手几个问题，了解主流消息队列RabbitMq,RocketMq,Kafka，整理好自己回顾用，顺便知识分享。

1.使用MQ原因

MQ的作用

MQ 可以实现异步处理

MQ 可以实现削峰填谷，使用它可以解决短时间内爆发式的请求任务，在不使用 MQ 的情况下会导致服务处理不过来，出现应用程序假死的情况，使用了 MQ 之后可以把这些请求先暂存到消息队列中，然后进行排队执行

MQ 可以实现 对日志的采集和转发，比如有多个日志写入到程序中，然后把日志添加到 MQ，紧接着由日志处理系统订阅 MQ，最后 MQ 将消息接收并转发给日志处理系统

MQ的问题

增加了系统的运行风险

引入 MQ 系统，则意味着新增了一套系统，并且其他的业务系统会对 MQ 系统进行深度依赖，系统部署的越多则意味着发生故障的可能性就越大，如果 MQ 系统挂掉的话可能会导致整个业务系统瘫痪

增加了系统的复杂度

引入 MQ 系统，需要考虑消息丢失、消息重复消费、消息的顺序消费等问题，同时还需要引入新的客户端来处理 MQ 的业务，增加了编程的运维门槛，增加了系统的复杂性

2.前世今生

RabbitMq

RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。

RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。该公司在2010年4月被SpringSource（VMWare的一个部门）收购。在2013年5月被并入Pivotal。其实VMWare，Pivotal和EMC本质上是一家的。不同的是VMWare是独立上市子公司，而Pivotal是整合了EMC的某些资源，现在并没有上市。

RocketMq

RocketMQ是阿里巴巴研发的一款纯Java开发的分布式、高性能、高可靠、高实时的消息中间件。2016年11⽉28⽇，阿⾥巴巴向 Apache 软件基⾦会捐赠 RocketMQ，成为 Apache 孵化项⽬，并在2017 年 9 ⽉ 25 ⽇正式成为 Apache 顶 级项⽬，成为国内⾸个互联⽹中间件在 Apache 上的顶级项⽬。

RocketMQ广泛应用于阿里巴巴内部的生产系统，满足线上海量消息堆积的需求。经历多年双十一的洗礼，在可用性、可靠性和稳定性方面都有着非常稳定的表现，证明了其是一款非常优秀的消息中间件。

• 2011年初Kafka开源后，淘宝中间件团队在对Kafka进行了深入研究后，开发了一款MetaQ。
• 2012年，MetaQ发展到了v3.0版本，并在它基础上进行了进一步的抽象，形成了RocketMQ，进行了开源。
• 2016年双十一，RocketMQ承载了万亿级消息的流转，跨越了一个新的里程碑。
• 2016 年 11⽉28⽇，阿⾥巴巴 向 Apache 软件基⾦会捐赠 RocketMQ，成为 Apache 孵化项⽬。
• 2017 年 9 ⽉ 25 ⽇，Apache 宣布 RocketMQ孵化成为 Apache 顶级项⽬，成为国内首个互联网中间件在 Apache 上的顶级项⽬。

Kafka

kafka的架构师jay kreps对于kafka的名称由来是这样讲的，由于jay kreps非常喜欢franz kafka,并且觉得kafka这个名字很酷，因此取了个和消息传递系统完全不相干的名称kafka，该名字并没有特别的含

kafka的诞生，是为了解决linkedin的数据管道问题，起初linkedin采用了ActiveMQ来进行数据交换，大约是在2010年前后，那时的ActiveMQ还远远无法满足linkedin对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，linkedin决定研发自己的消息传递系统，当时linkedin的首席架构师jay kreps便开始组织团队进行消息传递系统的研发；

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

3.基础知识

RabbitMq

使用场景

开源项目，社区活跃，中小型工时使用

• RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。
• AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
• AMQP协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。
• RabbitMq比kafka成熟，在可用性上，稳定性上，可靠性上，RabbitMq超过kafka

优点

吞吐量高，功能齐全

管理界面易用

社区活跃，性能极好，；

缺点

吞吐量只是万级

erlang难以二次开发和掌控

集群动态扩展非常麻烦

RocketMq

使用场景

大型公司，基础研发能力强，使用rocketMq

• RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。
• RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，
• RocketMQ在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

优点

单机吞吐量：十万级
可用性：非常高，分布式架构
消息可靠性：经过参数优化配置，消息可以做到0丢失
功能支持：MQ功能较为完善，还是分布式的，扩展性好
支持定时消息；
支持消费失败重试；
支持10亿级别的消息堆积，不会因为堆积导致性能下降
源码是java，我们可以自己阅读源码，定制自己公司的MQ，可以掌控

缺点

支持的客户端语言不多，目前是java及c++，其中c++不成熟；

社区活跃度一般

没有在 mq 核心中去实现JMS等接口，有些系统要迁移需要修改大量代码

Kafka

使用场景

大数据领域实时计算，日志采集，用kafka是业内标准

• Kafka是linkedin开源的MQ系统，主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输，0.8开始支持复制，不支持事务，适合产生大量数据的互联网服务的数据收集业务。
• Kafka设计的初衷就是处理日志的，可以看做是一个日志系统，针对性很强，所以它并没有具备一个成熟MQ应该具备的特性
• 为大数据而生的消息中间件，以其百万级TPS的吞吐量名声大噪，迅速成为大数据领域的宠儿，在数据采集、传输、存储的过程中发挥着举足轻重的作用

优点

kafka高吞吐率的实现：
1.顺序读写：kafka将消息读写写入到了分区partition中，而分区消息是顺序读写的。顺序读写要远快于随机读写
2.零拷贝：生产者、消费者对于kafka中消息的操作都是采用零拷贝实现的
3.批量发送：kafka允许采用批量消息发送模式
4.消息压缩：kafka允许对消息集合进行压缩

性能卓越，单机写入TPS约在百万条/秒，最大的优点，就是吞吐量高。
时效性：ms级
可用性：非常高，kafka是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用
消费者采用Pull方式获取消息, 消息有序, 通过控制能够保证所有消息被消费且仅被消费一次
有优秀的第三方Kafka Web管理界面Kafka-Manager；
在日志领域比较成熟，被多家公司和多个开源项目使用；

缺点

Kafka单机超过64个队列/分区，Load会发生明显的飙高现象，队列越多，load越高，发送消息响应时间变长

使用短轮询方式，实时性取决于轮询间隔时间

消费失败不支持重试

不支持定时消息

支持消息顺序，但是一台代理宕机后，就会产生消息乱序； 社区更新较慢

	rabbitMq	rocketMq	kafka
单机吞吐量	万级	十万级	十万级
topic数量对吞吐量的影响		几百几千topic影响很小	topic变多吞吐量下降
时效性	微秒	ms	ms以内
可用性	高，主从高可用	非常高，分布式
消息可靠性		经过参数配置0丢失
功能支持	基于erlang开发，并发能力强，性能好，延迟低	性能较为完善，分布式，扩展性好	大数据实时计算和日志采集方面应用较多

4.保证消息处理顺序

1.为什么要保证顺序

消息队列中的若干消息如果是对同一个数据进行操作，这些操作具有前后的关系，必须要按前后的顺序执行，否则就会造成数据异常。

RabbitMq

乱序原因

①一个queue，有多个consumer去消费，这样就会造成顺序的错误，consumer从MQ里面读取数据是有序的，但是每个consumer的执行时间是不固定的，无法保证先读到消息的consumer一定先完成操作，这样就会出现消息并没有按照顺序执行，造成数据顺序错误。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

②一个queue对应一个consumer，但是consumer里面进行了多线程消费，这样也会造成消息消费顺序错误。

解决方案

①拆分多个queue，每个queue一个consumer，就是多一些queue而已，确实是麻烦点；这样也会造成吞吐量下降，可以在消费者内部采用多线程的方式取消费。

②或者就一个queue但是对应一个consumer，然后这个consumer内部用内存队列做排队，然后分发给底层不同的worker来处理

RocketMq

乱序原因

Broker中的每个Topic都有多个Queue，写入消息的时候会平均分配（负载均衡机制，默认轮询，也可以自定义）给不同的Queue，假如我们有一个消费者组ComsumerGroup，这个消费组中的每一台机器都会负责一部分Queue，那么就会导致顺序的乱序问题

解决方案

1. 保证Producer、Queue、Comsumer是一对一对一的关系

2. 把需要保持顺序消费的消息放到同一个Queue中，且让同一台机子处理

Kafka

乱序原因

①kafka一个topic，一个partition，一个consumer，但是consumer内部进行多线程消费，这样数据也会出现顺序错乱问题。

②具有顺序的数据写入到了不同的partition里面，不同的消费者去消费，但是每个consumer的执行时间是不固定的，无法保证先读到消息的consumer一定先完成操作，这样就会出现消息并没有按照顺序执行，造成数据顺序错误。

解决方案

①确保同一个消息发送到同一个partition，一个topic，一个partition，一个consumer，内部单线程消费。

②写N个内存queue，然后N个线程分别消费一个内存queue即可

5.消息不丢失原理

原理总结

生产者

确认机制

队列本身

刷盘

消费者

手动ACK

RabbitMq

Producer保证消息不丢失

1、rabbitMQ引入了事务机制和确认机制（confirm）

• 确认机制，是当mq收到生产者发送的消息时，会返回一个ack告知生产者，收到了这条消息，如果没有收到，那就采取重试机制后者其他方式补偿。

• 事务机制开启之后，相当于同步执行，必然会降低系统的性能，一般我们不采用这种方式。

2、重试机制

rabbitmq为生产者设置了重试机制默认是3次，可以修改重试次数，超过了最大重试次数限制采取人工补偿机制。

Broker保证消息不丢失

1、rabbitMq持久化机制

• 消息到达mq之后，mq宕机了，然后消息又没有进行持久化，这时消息就会丢失。
• 开启mq的持久化机制，消息队列，交换机、消息都要开启持久化。

2、如果队列满了，多余的消息发送到Broker时可以使用死信队列保证消息不会被丢弃

Consumer保证消息不丢失

1.开启消费端的手动ack

2、可以使用消费者的重试机制，重试超过最大次数还没成功则采取人工补偿机制。

RocketMq

Producer保证消息不丢失

RocketMQ发送消息有三种模式，同步发送，异步发送、单向发送。

• 同步发送消息时会同步阻塞等待Broker返回发送结果，如果发送失败不会收到发送结果SendResult,这种是最可靠的发送方式。
• 异步发送消息可以在回调方法中得知发送结果。
• 单向发送是消息发送完之后就不管了，不管发送成功没成功，是最不可靠的一种方式。

mq为生产者提供了失败重试机制，同步发送和异步发送默认都是失败重试两次当然可以修改重试次数，如果多次还是失败，那么可以采取记录这条信息，然后人工采取补偿机制。

Broker保证消息不丢失

1、刷盘策略

RocketMq持久化消息有两种策略即同步刷盘和异步刷盘。

• 默认情况下是异步刷盘，此模式下当生产者把消息发送到broker，消息存到内存之后就认为消息发送成功了，就会返回给生产者消息发送成功的结果。但是如果消息还没持久化到硬盘，服务器宕机了，那么消息就会丢失。

• 同步刷盘是当Broker接收到消息并且持久化到硬盘之后才会返回消息发送成功的结果，这样就会保证消息不会丢失，但是同步刷盘相对于异步刷盘来说效率上有所降低，大概降低10%，具体情况根据业务需求设定吧。

• 修改配置文件中刷盘方式，ASYNC_FLUSH=异步刷盘，SYNC_FLUSH=同步刷盘

2、集群模式

rocketmq的集群模式保证可rocketMQ高可用。利用多master多slave节点保证rocketmq的高可用。

此模式是broker保证消息不丢失的配置，主从复制同步复制，刷盘模式同步刷盘，但是这种模式下性能会有所降低。

Consumer保证消息不丢失

1、手动ack

消费端消费成功之后，手动确认删除队列的消息

2、消费者消费失败重试机制

消费者消费失败会自动重试，如果消费失败没有手动ack则会自动重试15次。

Kafka

Producer保证消息不丢失

1、producer的ack机制

kafka的生产者确认机制有三种取值分别为0、1、-1（all）

acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障）。
acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有follwer服务器的完全确认即可做出回应，在这种情况下，当leader还没有将数据同步到Follwer宕机，存在丢失数据的可能性。
acks = -1代表所有的所有的分区副本备份完成，不会丢失数据这是最强有力的保证。但是这种模式往往效率相对较低。
2、producer重试机制

Broker保证消息不丢失

kafka的broker使用副本机制保证数据的可靠性。每个broker中的partition我们一般都会设置有replication（副本）的个数，生产者写入的时候首先根据分发策略（有partition按partition，有key按key，都没有轮询）写入到leader中，follower（副本）再跟leader同步数据，这样有了备份，也可以保证消息数据的不丢失。

Consumer保证消息不丢失

1、手动ack

2、offset commit

消费者通过offset commit 来保证数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，会接着上次的offset进行消费。kafka并不像其他消息队列，消费完消息之后，会将数据从队列中删除，而是维护了一个日志文件，通过时间和储存大小进行日志删除策略。如果offset没有提交，程序启动之后，会从上次消费的位置继续消费，有可能存在重复消费的情况。

Offset Reset 三种模式

earliest（最早）：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费。
latest(最新的)：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据。
数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，会接着上次的offset进行消费。kafka并不像其他消息队列，消费完消息之后，会将数据从队列中删除，而是维护了一个日志文件，通过时间和储存大小进行日志删除策略。如果offset没有提交，程序启动之后，会从上次消费的位置继续消费，有可能存在重复消费的情况。

Offset Reset 三种模式

earliest（最早）：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费。
latest(最新的)：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据。
none(没有)：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。