java—12 kafka

一、消息队列的优缺点

二、常用MQ

1. Kafka

2. RocketMQ

3. RabbitMQ

4. ActiveMQ

5. ZeroMQ

6. MQ选型对比

适用场景——从公司基础建设力量角度出发

适用场景——从业务场景角度出发

四、基本概念和操作

1. kafka常用术语

2. kafka常用指令

3. 单播消息（group.id相同）

4. 多播消息（group.id不相同）

5. 消费组

编辑6. 稀疏索引

五、kafka集群

1. 集群搭建

2. 集群启动和验证

3. Topic的意义

4. Topic和Partition

5. 分区

6. 副本

7. 集群操作指令

8. 多分区&多副本

9. 多分区消费组

10. Controller

11. Rebalance机制

12. HW和LEO

六、高频面试题

1. 如何防止消息丢失？

2. 如何防止消息的重复消费?

3. 如何做到顺序消费?

4. 如何解决消息积压的问题?

5. 如何实现延迟队列?

6. Kafka如何做到单机上百万的高吞吐量呢？

7. Kafka高吞吐——非零拷贝技术

8. Kafka高吞吐——零拷贝技术

一、消息队列的优缺点

消息队列的优点：

① 实现系统解耦
② 实现异步调用
③ 流量削峰

消息队列的缺点：

① 系统可用性降低
② 提升系统的复杂度
③ 数据一致性问题

二、常用MQ

1. Kafka

Apache Kafka是一个分布式消息发布订阅系统。它最初由LinkedIn公司基于独特的设计实现为一个分布式的日志提交系统(a distributed commit log)，之后成为Apache项目的一部分。Kafka性能高效、可扩展良好并且可持久化。它的分区特性，可复制和可容错都是其不错的特性。

优点：客户端语言丰富：支持Java、.Net、PHP、Ruby、Python、Go等多种语言高性能：单机写入TPS约在100万条/秒，消息大小10个字节；提供完全分布式架构，并有replica机制，拥有较高的可用性和可靠性，理论上支持消息无限堆积；消费者采用Pull方式获取消息。消息有序，通过控制能够保证所有消息被消费且仅被消费一次；在日志领域比较成熟，被多家公司和多个开源项目使用。有管理界面Kafka-Manager；

缺点： Kafka单机超过64个队列/分区时，Load时会发生明显的飙高现象。队列越多，负载越高，发送消息响应时间变长；使用短轮询方式，实时性取决于轮询间隔时间；消费失败不支持重试；社区更新较慢。

2. RocketMQ

RocketMQ出自阿里的开源产品，用Java语言实现，在设计时参考了Kafka，并做出了自己的一些改进，消息可靠性上比Kafka更好。RocketMQ在阿里内部被广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。

优点：单机支持1万以上持久化队列； RocketMQ的所有消息都是持久化的，先写入系统PAGECACHE，然后刷盘，可以保证内存与磁盘都有一份数据，而访问时，直接从内存读取。模型简单，接口易用（JMS的接口很多场合并不太实用）；性能非常好，可以允许大量堆积消息在Broker中；支持多种消费模式，包括集群消费、广播消费等；各个环节分布式扩展设计，支持主从和高可用；开发度较活跃，版本更新很快。

缺点：支持的客户端语言不多，目前是Java及C++，其中C++还不成熟； RocketMQ社区关注度及成熟度也不及Kafka；没有Web管理界面，提供了一个 CLI (命令行界面) 管理工具带来查询、管理和诊断各种问题；没有在MQ核心里实现JMS等接口；

3. RabbitMQ

RabbitMQ于2007年发布，是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。它提供了多种技术可以让你在性能和可靠性之间进行权衡。这些技术包括持久性机制、投递确认、发布者证实和高可用性机制；灵活的路由，消息在到达队列前是通过交换机进行路由的。RabbitMQ为典型的路由逻辑提供了多种内置交换机类型。

优点：由于Erlang语言的特性，消息队列性能较好，支持高并发；健壮、稳定、易用、跨平台、支持多种语言、文档齐全；有消息确认机制和持久化机制，可靠性高；高度可定制的路由；管理界面较丰富，在互联网公司也有较大规模的应用，社区活跃度高。

缺点：实现了代理架构，意味着消息在发送到客户端之前可以在中央节点上排队。此特性使得RabbitMQ易于使用和部署，但是使得其运行速度较慢，因为中央节点增加了延迟，消息封装后也比较大；需要学习比较复杂的接口和协议，学习和维护成本较高。尽管结合 Erlang 语言本身的并发优势，性能较好，但是不利于做二次开发和维护；

4. ActiveMQ

ActiveMQ是由Apache出品，ActiveMQ是一个完全支持JMS1.1和J2EE 1.4规范的JMS Provider实现。它非常快速，支持多种语言的客户端和协议，而且可以非常容易的嵌入到企业的应用环境中，并有许多高级功能。

优点：可以用JDBC：可以将数据持久化到数据库。虽然使用JDBC会降低ActiveMQ的性能，但是数据库一直都是开发人员最熟悉的存储介质；支持JMS规范：支持JMS规范提供的统一接口; 支持自动重连和错误重试机制；有安全机制：支持基于shiro，jaas等多种安全配置机制，可以对Queue/Topic进行认证和授权；监控完善：拥有完善的监控，包括WebConsole，JMX，Shell命令行，Jolokia的RESTful API；界面友善：提供的WebConsole可以满足大部分情况，还有很多第三方的组件可以使用，比如hawtio；

缺点：社区活跃度不及RabbitMQ高；根据其他用户反馈，会出莫名其妙的问题，会丢失消息；目前重心放到activemq6.0产品Apollo，对5.x的维护较少；不适合用于上千个队列的应用场景；

5. ZeroMQ

号称史上最快的消息队列，它实际类似于Socket的一系列接口，他跟Socket的区别是：普通的socket是端到端的（1:1的关系），而ZMQ却是可以N：M 的关系，人们对BSD套接字的了解较多的是点对点的连接，点对点连接需要显式地建立连接、销毁连接、选择协议（TCP/UDP）和处理错误等，而ZMQ屏蔽了这些细节，让你的网络编程更为简单。ZMQ用于node与node间的通信，node可以是主机或者是进程。

引用官方的说法： “ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层，像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ的明确目标是“成为标准网络协议栈的一部分，之后进入Linux内核”。

与RabbitMQ相比，ZMQ并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，更像一个底层的网络通讯库，在Socket API之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的API接口。支持“Request-Reply “，”Publisher Subscriber“，”Parallel Pipeline”三种基本模型和扩展模型。

6. MQ选型对比

总结：消息队列的选型需要根据具体应用需求而定，ZeroMQ小而美，RabbitMQ大而稳，Kakfa和RocketMQ快而强劲。

适用场景——从公司基础建设力量角度出发

中小型软件公司，建议选RabbitMQ 首先：erlang语言天生具备高并发的特性，而且他的管理界面用起来十分方便。他的弊端也很明显，虽然RabbitMQ是开源的，然而国内有几个能定制化开发erlang的程序员呢？所幸， RabbitMQ的社区十分活跃，可以解决开发过程中遇到的bug，这点对于中小型公司来说十分重要。

其次：不考虑Kafka的原因是，一方面中小型软件公司不如互联网公司，数据量没那么大，选消息中间件，应首选功能比较完备的，所以kafka排除。最后：不考虑RocketMQ的原因是，RocketMQ是阿里出品，如果阿里放弃维护RocketMQ，中小型公司一般抽不出人来进行RocketMQ的定制化开发，因此不推荐。

大型软件公司根据具体使用在RocketMQ和kafka之间二选一。一方面，大型软件公司，具备足够的资金搭建分布式环境，也具备足够大的数据量。针对RocketMQ，大型软件公司也可以抽出人手对RocketMQ进行定制化开发，毕竟国内有能力改JAVA源码的人，还是相当多的。至于kafka，根据业务场景选择，如果有日志采集功能，肯定是首选kafka了。

适用场景——从业务场景角度出发

RocketMQ定位于非日志的可靠消息传输（日志场景也OK），目前RocketMQ在阿里集团被广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。

Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。

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

~~三、安装~~

四、基本概念和操作

1. kafka常用术语

名词	解释
Broker	【节点】一个Kafka节点就是一个Broker，一个和多个Broker可以组成一个Kafka集群
Topic	【主题】Kafka根据topic对消息进行归类，发布到kafka集群的每套消息都需要指定一个topic，topic是一个逻辑概念，物理上是不存在的
Producer	【生产者】用于向Kafka中发送消息
Consumer	【消费者】从Kafka中获取消息
Consumer Group	【消费组】每个Consumer都会归属于一个消费组，一条消息可以同时被多个不同的消费组消费，但是只能被一个消费组中的消费者消费
Partition	【分片】物理上的概念，可以将一个topic上的数据拆分为多分放到 Partition中，每个Patition内部的消息是有序的。

2. kafka常用指令

创建主题

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic muse -- partitions 1 --replication-factor 1

查看主题

./kafka-topics.sh --list --bootstrap-server localhost:9092

开启消息生产端

./kafka-console-producer.sh --topic muse --bootstrap-server localhost:9092

开启消息消费端

./kafka-console-consumer.sh --topic muse --bootstrap-server localhost:9092
./kafka-console-consumer.sh --topic muse --bootstrap-server localhost:9092 -- from-beginning

消息是会被存储在kafka中的文件里的，并且是顺序存储的，消息有偏移量的概念，所以我们可以指定偏移量去读取某个位置的消息。

3. 单播消息（group.id相同）

一个消费组里，只会有一个消费者能够消费到某个topic中的消息。

首先，打开两个窗口，分别执行如下语句开启消费端，那么就在“museGroup1”消费组中创建了两个Consumer

./kafka-console-consumer.sh --topic muse --bootstrap-server localhost:9092 -- consumer-property group.id=museGroup1

然后：Producer端发送3条消息，我们发现，只有一个Consumer收到了消息。

4. 多播消息（group.id不相同）

当业务场景中，需要同一个topic下的消息被多个消费者消费，那么我们可以采用创建多个消费组的方式，那么这种方式就是多播消息。打开两个窗口，分别执行如下指令：

./kafka-console-consumer.sh --topic muse --bootstrap-server localhost:9092 --consumer property group.id=museGroup1
./kafka-console-consumer.sh --topic muse --bootstrap-server localhost:9092 --consumer property group.id=museGroup2

最后，Producer端发送3条消息，我们发现，两个Consumer都收到了消息。如下所示：

5. 消费组

查看当前主题下有哪些消费组

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

查看消费组中的具体信息

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group museGroup1 --describe

其中，展现出来的信息有如下含义：

CURRENT-OFFSET：当前消费组已消费消息的偏移量。
LOG-END-OFFSET：主题对应分区消息的结束偏移量（水位——HW）。
LAG：当前消费组堆积未消费的消息数量。

__consumer_offsets-N

kafka默认创建了一个拥有50个分区的主题，名称为：“__consumer_offsets”。

consumer会将消费分区的offset提交给kafka内部topic：__consumer_offsets，提交过去的时候，【key】=consumerGroupID+topic+分区号，【value】=当前offset的值。

Kafka会定期清理topic里的消息，最后就保留最新的那条数据。可以通过如下公式确定 consumer消费的offset要提交到哪个__consumer_offsets hash(consumerGroupID)% 主题“__consumer_offsets”的分区数量

6. 稀疏索引

五、kafka集群

1. 集群搭建

创建kafka-cluster目录，并解压kafka_2.13-3.0.0.tgz为3份kafka

修改kafka1的server.propertis配置文件

broker.id=10 listeners=PLAINTEXT://localhost:9093 log.dirs=/Users/muse/Lesson/ServerContext/kafka-cluster/kafka1/kafka-logs

修改kafka2的server.propertis配置文件

broker.id=11 listeners=PLAINTEXT://localhost:9094 log.dirs=/Users/muse/Lesson/ServerContext/kafka-cluster/kafka2/kafka-logs

修改kafka3的server.propertis配置文件

broker.id=12 listeners=PLAINTEXT://localhost:9095 log.dirs=/Users/muse/Lesson/ServerContext/kafka-cluster/kafka3/kafka-logs

2. 集群启动和验证

启动kafka1，kafka2和kafka3

./kafka1/bin/kafka-server-start.sh -daemon ./kafka1/config/server.properties
./kafka2/bin/kafka-server-start.sh -daemon ./kafka2/config/server.properties
./kafka3/bin/kafka-server-start.sh -daemon ./kafka3/config/server.properties

验证3个Broker是否启动成功

首先，可以通过ps -ef | grep kafka来查看进程是否启动
其次：在zookeeper中查看/brokers/ids下中是否有相应的brokerId目录生成

3. Topic的意义

试想一下，当我们要尝试发送 /消费消息的时候需要注意什么呢？“这有啥需要注意的，发送不就得了！” 结果，我们发现了一个非常重大的问题，大家都往Kafka中发送消息，所有的消息都混合在了一起，就类似所有快递公司的快递员（Producer端）把快递都扔到了一个大仓库里，结果，去取快递的小伙伴们（Consumer端）面对堆积如山且混乱不堪的“快递山”—— 疯了。。。

4. Topic和Partition

5. 分区

一个主题中的消息量是非常大的，因此可以通过分区的设置，来分布式存储这些消息。并且分区也可以提供消息并发存储的能力。

6. 副本

如果分片挂掉了，数据就丢失了。那么为了提高系统的可用性。我们把分片复制多个，这就是副本了。但是，副本的产生，也会随之带来数据一致性的问题，即：有的副本写数据成功，但是有的副本写数据失败。

Leader：kafka的读写操作都发生在leader上，leader负责把数据同步给follower。当leader挂掉了，那么经过主从选举，从多个follower中选举产生一个新的leader。

Follower：follower接收leader同步过来的数据，它不提供读写（主要是为了保证多副本数据与消费的一致性）

7. 集群操作指令

为名称为muse-rp的Topic创建2个分区（--partitions）3个副本（--replication-factor）

./kafka-topics.sh --create --topic muse-rp --bootstrap-server localhost:9093 localhost:9094 localhost:9095 --partitions 2 --replication-factor 3

删除Topic

./kafka-topics.sh --delete --topic muse-rp --bootstrap-server localhost:9093 localhost:9094 localhost:9095

查看分区和副本分布情况

./kafka-topics.sh --describe --topic muse-rp --bootstrap-server localhost:9093 localhost:9094 localhost:9095

消息发送和消费

./kafka-console-producer.sh --topic muse-rp --bootstrap-server localhost:9093 localhost:9094 localhost:9095
./kafka-console-consumer.sh --topic muse-rp --bootstrap-server localhost:9093 localhost:9094 localhost:9095 --consumer-property group.id=museGroup1

8. 多分区&多副本

为名称为muse-rp的Topic创建---------------------------- 2个分区（--partitions） 3个副本（--replication-factor）

9. 多分区消费组

一个partition只能被一个消费组中的一个消费者消费，这样设计的目的是保证消息的有序性，但是在多个partition的多个消费者消费的总顺序性是无法得到保证的。

partition的数量决定了消费组中Consumer的数量，建议同一个消费组中的Consumer数量不要超过partition的数量，否则多余的Consumer就无法消费到消息了。

但是，如果消费者挂掉了，那么就会触发rebalance机制，会由其他消费者来消费该分区。

10. Controller

Kafka集群中的Broker在ZK中创建临时序号节点，序号最小的节点也就是最先创建的那个节点，将作为集群的Controller，负责管理整个集群中的所有分区和副本的状态。

Controller控制器的作用如下：

① 当某个分区的leader副本出现故障时，由控制器负责为该分区选举出新的leader副本。
② 当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
③ 当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

11. Rebalance机制

当消费者没有指明分区消费时，消费组里的消费者和分区关系发生了变化，那么就会触发 rebalance机制。这个机制会重新调整消费者消费哪个分区。

在触发rebalance机制之前，消费者消费哪个分区有3种策略：

1> range：通过公式来计算某个消费者消费哪个分区。

2> 轮询：大家轮流对分区进行消费。

3> sticky：在触发rebalance之后，在消费者消费的原分区不变的基础上进行调整。

12. HW和LEO

HW（HighWatermark）俗称高水位，取一个partition对应的ISR中最小的LEO（log-end offset）作为HW；

Consumer最多只能消费到HW所在的位置，每个副本都有HW，Leader和Follower各自负责更新自己的HW的状态。

对于Leader新写入的消息，Consumer不能立刻消费，Leader会等待该消息被所有ISR中的副本同步后更新HW，此时消息才能被Consumer所消费。这样就能保证如果Leader所在的 broker失效，该消息仍然可以从新选举的Leader中获取。

具体逻辑请看下一张图：

六、高频面试题

1. 如何防止消息丢失？

针对发送方：将ack设置为1或者-1/all可以防止消息丢失，如果要做到99.9999%，ack要设置为all，并把min.insync.replicas配置成分区备份数

针对接收方：把自动提交修改为手动提交(enable-auto-commit)。

2. 如何防止消息的重复消费?

一条消息被消费者消费多次，如果为了不消费到重复的消息，我们需要在消费端增加幂等性处理，例如：

① 通过mysql插入业务id作为主键，因为主键具有唯一性，所以一次只能插入一条业务数据。
② 使用redis或zk的分布式锁，实现对业务数据的幂等操作。

3. 如何做到顺序消费?

针对发送方：在发送时将ack配置为非0，确保消息至少同步到leader之后再返回ack继续发送。但是，只能保证分区内部的消息是顺序的，而无法保证一个Topic下的多个分区总的消息是有序的。

针对接收方：消息发送到一个分区中，只配置一个消费组的消费者来接收消息，那么这个 Consumer所接收到的消息就是有顺序的了，不过这也就牺牲掉了性能。

4. 如何解决消息积压的问题?

消息积压会导致很多问题，比如：磁盘被打满、Producer发送消息导致kafka 性能过慢，然后就有可能发生服务雪崩。解决的方案如下所示：

① 提升一个Consumer的处理能力。即：在一个消费者中启动多个线程，让多线程加快消费的速度。

② 提升总体Consumer的处理能力。启动多个消费组，增加Consumer的数量从而提高消费能力。

③ 如果业务运行，设定某个时间内，如果消息仍没有被消费，那么Consumer收到消息后，直接废弃掉，不执行下面的业务逻辑。

5. 如何实现延迟队列?

应用场景：订单创建成功后如果超过30分钟没有付款，则需要取消订单。

① 创建一个表示“订单30分钟未支付”的Topic，如：order_not_paid_30min，表示延迟30分钟的消息队列。
② Producer发送消息的时候，消息内容要带上订单生成的时间create_time。
③ Consumer消费Topic中的消息，如果发现now减去create_time不足30分钟，则不去消费；记录当前的offset，不去消费当前以及之后的消息。
④ 通过记录的offset去获取消息，如果发现消息已经超过30分钟且订单状态是“未支付”，那么则将订单状态设置为“取消”，然后获取下一个offset的消息。

6. Kafka如何做到单机上百万的高吞吐量呢？

写入数据：主要是依靠页面缓存技术 + 磁盘顺序写实现的。
读取数据：主要依靠零拷贝技术实现的。

7. Kafka高吞吐——非零拷贝技术

通过下图过程的描述，很明显可以看到存在两次没必要的copy，一次是从OS Cache里拷贝到Kafka进程的缓存里，接着又从Kafka进程的缓存里拷贝回OS的Socket缓存里。而且为了进行这两次拷贝，中间还发生了好几次上下文切换，一会儿是Kafka进程在执行，一会儿上下文切换到操作系统来执行，所以这种方式来读取数据是比较消耗性能的。

8. Kafka高吞吐——零拷贝技术

Kafka为了解决非零拷贝这个问题，在读数据的时候是引入零拷贝技术。也就是说，通过 OS的sendfile技术直接让OS Cache中的数据发送到网卡后传输给下游的消费者，中间跳过了两次拷贝数据的步骤。通过零拷贝技术，就不需要把OS Cache里的数据拷贝到应用缓存，再从应用缓存拷贝到 Socket 缓存了，两次拷贝都省略了，所以叫做零拷贝