首页  思维导图  详情

Kafka消息引擎

2021-02-20 15:37:44   3  举报





AI智能生成

为你推荐

查看更多



Kafka

rocketMQ与kafka的消息发送流程

消息

消息中心

消息接收端消息设计

全面介绍Kafka消息引擎，主体大纲从技术简介、发展历程、专业术语、版本选型、部署方案、核心参数、技术原理、高级功能、集群监控、性能调优、最佳实践、Kafka Streams、Apache Kafka等方面解析Kafka。

消息引擎

Kafka Streams

性能调优

消息队列

Kafka技术栈

作者其他创作

大纲/内容

Apache Kafka是消息引擎系统，也是一个分布式流处理平台（Distributed Streaming Platform）

技术简介

数据正确性不足。因为数据的收集主要采用轮询的方式，如何确定轮询的间隔时间就变成一个高度经验化的事情，虽然可以采用一些类似启发式算法（Heuristic）来帮助评估间隔时间值，但一旦指定不当，必然会造成较大的数据偏差。

系统高度定制化，维护成本高。各个业务子系统都需要对接数据收集模块，引入了大量的定制开销和人工成本。

LinkedIn最初碰到的历史问题？

因为Kafka系统的写性能很强，所以找了个作家的名字来命名似乎是个好主意。大学期间我上了很多文学课，非常喜欢Franz Kafka这个作家，另外开源软件起这个名字听上去很酷。

Kafka名字的由来？

提供一套API实现生产者和消费者

降低网络传输和磁盘存储开销

实现高伸缩架构

提供三个方面的特性

2011年Kafka正式进入到Apache基金会孵化并于次年10月顺利毕业成为Apache顶级项目

正式开源

发展历程

为了解决这些问题，尝试使用ActiveMQ来解决，效果不理想，随后就诞生了Kafka

Kafka处理的主要对象

消息：Record

主题是承载消息的逻辑容器，实际使用中多用来区分具体业务

主题：Topic

一个有序不变的消息序列，每个主题下可以有多个分区

分区：Partition

表示分区中每条消息的位置信息，是一个单调递增且不变的值

消息位移：Offset

Kafka中同一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者副本和追随副本，各自有不同的角色划分。副本是在分区层级下的，即每个分区可以配置多个副本实现高可用。

副本：Replica

向主题发布新消息的应用程序

生产者：Producer

从主题订阅新消息的应用程序

消费者：Consumer

表征消费者消费进度，每个消费者都有自己的消费者位移

消费者位移：Consumer Offset

多个消费者实例共同组成一个组，同时消费多个分区以实现高吞吐

消费者组：Consumer Group

消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程，Rebalance是Kafka消费者端实现高可用的重要手段

重平衡：Rebalance

专业术语

简介：社区版Kafka，开发人数最多，版本迭代速度最快，仅提供最基础的组件，Kafka Connect只提供一种连接器，即读写磁盘文件的连接器，没提供任何监控框架和工具（只能借助第三方监控Kafka Manager）

优势：迭代速度快，社区响应度高，有更高的把控度

劣势：仅提供核心基础组件，缺失一些高级特性

建议：如果仅仅只需要一个消息引擎系统亦或是简单的流处理应用场景，同时需要对系统有较大把控度，推荐使用此版本

Apache Kafka

简介：分免费版和企业版两种。前者和Apache Kafka非常相像，除常规组件外，免费版还包括Schema注册中心和REST proxy两大功能。除此以外，免费版还包含了更多的连接器，都是Confluent公司开发并认证过的。缺陷在于没有国内业务发展计划

优势：集成了很多高级特性且由Kafka原班人马打造，质量上有保证。

劣势：相关文档资料不全，国内普及效率低，没有太多可供参考的范例

建议：如果需要用到Kafka的一些高级特性，推荐使用此版本

Confluent Kafka

简介：大数据云公司发布的Kafka，天然集成Apache Kafka，有便捷的UI界面进行安装、运维、管理、监控

优势：操作简单，节省运维成本

劣势：把控度低，演进速度较慢

建议：如果需要快速搭建消息引擎系统，或者需要搭建的是多框架构成的数据平台且Kafka只是其中一个组件，那么推荐使用此版本。

CDH/HDP Kafka

版本选型

建议线上环境将Kafka部署在Linux操作系统，在以下3个方面，Linux的表现更胜一筹：1、I/O模型的使用 2、数据网络传输效率 3、社区支持度

Kafka客户端底层使用了Java的selecter，selecter在Linux上的实现机制是epoll，而在windows上的实现机制是select，因此在这一点上将Kafka部署在Linux上是有优势的，因为能够获得更高的I/O性能。

1、I/O模型的使用

Kafka需要在磁盘和网络间进行大量数据传输。在Linux部署Kafka能够享受到零拷贝技术所带来的快速数据传输性。

2、数据网络传输效率

社区目前对windows平台上发现的Kafka Bug不做任何承诺，一般不会修复。因此windows平台上部署Kafka只适合于个人测试或用于功能验证，切不要应用于生产环境。

3、社区支持度

建议：将Kafka部署在Linux上

考量点：操作系统I/O模型

总结

操作系统

1、应该选择普通的机械磁盘还是固态硬盘？

分支主题

使用RAID的优势在于：提供冗余的磁盘存储空间和提供负载均衡

2、到底是否应该使用磁盘阵列（RAID）？

建议：普通环境使用机械硬盘，不需要搭建RAID

考量点：磁盘I/O性能

磁盘

Kafka大量使用磁盘不假，但使用方式多是顺序读写操作，一定程度上规避了机械磁盘最大的劣势，即随机读写操作慢。从这一点来说，使用SSD似乎并没有太大的性能优势，而它易损坏而造成的可靠性差风缺陷，又由Kafka在软件层面提供机制来保证，所以使用普通机械磁盘是很划算的。

就Kafka而言，一方面Kafka自己实现了冗余机制来提高可靠性；另一方面通过分区的概念，Kafka也能在软件层面自行实现负载均衡。综合考虑：1、追求性价比的公司可以不搭建RAID，使用普通磁盘组成存储空间即可 2、使用机械磁盘完全能够胜任Kafka线上环境

磁盘容量规划需考虑以下几个元素：1、新增消息数 2、消息留存时间 3、平均消息大小 4、备份数 5、是否启用压缩

Kafka集群的存储容量规划？

每天消息总空间大小等于 1亿 * 1KB * 2 / 1000 / 1000 = 200GB 。Kafka集群除了消息数据还有其他类型比如所索引数据，再为这些数据预留 10% 的空间，总容量为 220 GB，保存两周，整体容量为 220GB * 14 ≈ 3TB 左右，Kafka支持数据压缩，假设压缩比例是0.75，那么最后需要规划的存储空间是 0.75 * 3 = 2.25TB

举例说明：假如每天需向Kafka集群发送1亿条消息，每条消息保存两份防止丢失，默认保存两周时间，假设消息平均大小是1KB，那么需预留多少磁盘空间？

建议：实际使用中建议预留20% ～ 30%的磁盘空间

考量点：根据消息数、存留时间预估磁盘容量

磁盘容量

带宽主要有两种：1Gbps的千兆网络和10Gbps的万兆网络，以千兆网络举例说明

如何进行宽带资源的规划？

由于带宽是1Gbps，即每秒处理1Gb的数据，假设每台Kafka服务器都是安装在专属的机器上，也就是说每台Kafka机器上没有混布其它服务，毕竟真实环境中不建议这么做，通常情况下，你只能假设Kafka会用到70%的带宽资源，因为总要为其它应用或进程留一些资源。

根据实际使用经验，超过70%的阈值就有网络丢包的可能性了，所以70%的设定是一个比较合理的值，也就是说单台Kafka服务器最多也就能使用700Mb的资源。这是它实用的最大宽带资源，你不能让Kafka服务器常规性使用这么多资源，故通常要再额外预留2/3的资源，即单台服务器使用带宽700Mb / 3 ≈ 240Mbps。（这里的2/3其实是相当保守的，你可以结合自己机器的使用情况酌情减少比值。）有了240Mbps，我们就可以计算1小时内处理1TB数据所需的服务器数量了。根据这个目标，我们每秒需要处理2336Mb（1024 * 1024 / 3600 * 8）的数据，除了240，约等于10台服务器。如果消息还需要额外复制2份，那么用的服务器数还需要乘以3，即30台。

举例说明：假设公司机房是千兆网络，即1Gbps，现在有个业务，其业务目标或SLA是在1小时内处理1TB的业务数据。那么问题来了，到底需要多少台Kafka服务器来完成这个业务？

对于千兆网络，建议每台服务器按照700Mbps来计算，避免大流量下的丢包

考量点：根据实际宽带资源和业务SLA预估服务器数量

带宽

部署方案

挂载在不同的物理磁盘好处：1、提升读写性能：比起单块磁盘，多块物理磁盘同时读写数据有更高的吞吐量。2、能够实现故障转移：即 Failover 。这是 Kafka 1.1 版本新引入的强大功能。要知道在以前，只要Kafka Broker 使用的任何一块磁盘挂掉了，整个 Broker 进程都会关闭。但是自 1.1 开始，这种情况被修正了，坏掉的磁盘上的数据会自动地转移到其他正常的磁盘上，而且 Broker 还能正常工作。

log.dirs

注意这是 dir ，结尾没有 s ，说明它只能表示单个路径，它是补充上一个参数用的

log.dir

存储信息

一般只需要设置 log.dirs，不用设置 log.dir

如果我让多个 Kafka 集群使用同一套 ZooKeeper 集群，那么这个参数应该怎么设置呢？这时候chroot 就派上用场了。这个 chroot 是 ZooKeeper 的概念，类似于别名。

zookeeper.connect

Zookeeper相关

学名叫监听器，其实就是告诉外部连接者要通过什么协议访问指定主机名和端口开放的 Kafka 服务

listeners

和 listeners 相比多了个 advertised 。 Advertised 的含义表示宣称的、公布的，就是说这组监听器是 Broker 用于对外发布的

advertised.listeners

你的 Kafka Broker 机器上配置了双网卡，一块网卡用于内网访问（即我们常说精选留言的内网 IP ）；另一个块用于外网访问。那么你可以配置 listeners 为内网 IP ， advertised.listeners为外网 IP

列出这两个参数就是想说你把它们忘掉吧，压根不要为它们指定值，毕竟都是过期的参数了

host.name/port

Broker 连接相关

是否允许自动创建 Topic

auto.create.topics.enable

建议最好设置成 false ，即不允许自动创建Topic

坚持之前的原则，坚决不能让那些落后太多的副本竞选 Leader 。这样做的后果是这个分区就不可用了，因为没有 Leader 了

false

那么 Kafka 允许你从那些 “ 跑得慢 ” 的副本中选一个出来当 Leader 。这样做的后果是数据有可能就丢失了，因为这些副本保存的数据本来就不全，当了 Leader 之后它本人就变得膨胀了，认为自己的数据才是权威的。

true

是否允许 Unclean Leader 选举

unclean.leader.election.enable

建议显式设置为false

如果设置为true表示允许 Kafka 定期地对一些 Topic 分区进行 Leader 重选举，当然这个重选举不是无脑进行的，它要满足一定的条件才会发生。严格来说它与上一个参数中 Leader 选举的最大不同在于，它不是选 Leader ，而是换 Leader ！比如 Leader A 一直表现得很好，但若 auto.leader.rebalance.enable=true ，那么有可能一段时间后 Leader A 就要被强行卸任换成Leader B

是否允许定期进行 Leader 选举

auto.leader.rebalance.enable

Topic 管理

换一次 Leader 代价很高的，原本向 A 发送请求的所有客户端都要切换成向 B 发送请求，而且这种换 Leader 本质上没有任何性能收益，因此我建议你在生产环境中把这个参数设置成false

这是个 “ 三兄弟 ” ，都是控制一条消息数据被保存多长时间。从优先级上来说 ms 设置最高、 minutes 次之、 hour 最低

log.retention.{hour|minutes|ms}

先说这个 “ 三兄弟 ” ，虽然 ms 设置有最高的优先级，但是通常情况下我们还是设置 hour 级别的多一些，比如 log.retention.hour=168 表示默认保存 7 天的数据，自动删除 7 天前的数据。很多公司把Kafka 当做存储来使用，那么这个值就要相应地调大。

这个值默认是 -1 ，表明你想在这台 Broker 上保存多少数据都可以，至少在容量方面 Broker 绝对为你开绿灯，不会做任何阻拦。

这个参数真正发挥作用的场景其实是在云上构建多租户的 Kafka 集群：设想你要做一个云上的 Kafka 服务，每个租户只能使用100GB 的磁盘空间，为了避免有个 “ 恶意 ” 租户使用过多的磁盘空间，设置这个参数就显得至关重要了

这是指定 Broker 为消息保存的总磁盘容量大小

log.retention.bytes

这个参数默认的977KB太少了，还不到 1MB

实际场景中突破 1MB 的消息都是屡见不鲜的，因此在线上环境中设置一个比较大的值还是比较保险的做法。毕竟它只是一个标尺而已，仅仅衡量 Broker 能够处理的最大消息大小，即使设置大一点也不会耗费什么磁盘空间的

控制 Broker 能够接收的最大消息大小

message.max.bytes

数据留存

Broker端参数

规定了该 Topic 消息被保存的时长。默认是 7 天，即该 Topic 只保存最近 7 天的消息。一旦设置了这个值，它会覆盖掉 Broker 端的全局参数值

retention.ms

规定了要为该 Topic 预留多大的磁盘空间。和全局参数作用相似，这个值通常在多租户的 Kafka 集群中会有用武之地。当前默认值是 -1 ，表示可以无限使用磁盘空间

retention.bytes

保存消息

它决定了 Kafka Broker 能够正常接收该 Topic 的最大消息大小

max.message.bytes

处理消息大小

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topictransaction --partitions 1 --replication-factor 1 --configretention.ms=15552000000 --configmax.message.bytes=5242880

需求：设想你的部门需要将交易数据发送到 Kafka 进行处理，需要保存最近半年的交易数据，同时这些数据很大，通常都有几 MB ，但一般不会超过5MB 。现在让我们用以下命令来创建 Topic ：

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-typetopics --entity-nametransaction --alter --add-config max.message.bytes=10485760

下面看看使用另一个自带的命令 kafka-configs 来修改 Topic 级别参数。假设我们现在要发送最大值是 10MB 的消息，该如何修改呢？命令如下：

创建 Topic 时进行设置

修改 Topic 时设置

设置Topic级别参数

Topic级别参数

将你的 JVM 堆大小设置成 6GB 吧，这是目前业界比较公认的一个合理值

堆大小（Heap Size）设置

如果 Broker 所在机器的 CPU 资源非常充裕，建议使用 CMS 收集器。启用方法是指定 -XX:+UseCurrentMarkSweepGC

否则，使用吞吐量收集器。开启方法是指定 -XX:+UseParallelGC

Java7/8

用默认的 G1 收集器就好了。在没有任何调优的情况下， G1 表现得要比 CMS 出色，主要体现在更少的 Full GC ，需要调整的参数更少等，所以使用G1 就好了

>=Java9

垃圾回收器（GC）设置

KAFKA_HEAP_OPTS：指定堆大小

KAFKA_JVM_PERFORMANCE_OPTS：指定GC参数

比如你可以这样启动 Kafka Broker ，即在启动 Kafka Broker 之前，先设置上这两个环境变量：$> export KAFKA_HEAP_OPTS=--Xms6g --Xmx6g$> export KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true$> bin/kafka-server-start.sh config/server.properties

Kafka相关

JVM参数

通常情况下将它设置成一个超大的值是合理的做法，比如ulimit -n 1000000

文件描述符限制

这里所说的文件系统指的是如ext3、ext4或XFS这样的日志型文件系统。根据官网的测试报告，XFS的性能要强于ext4，所以生产环境最好还是使用XFS。（ZFS性能更强劲）

文件系统类型

建议将swappniess配置成一个接近0但不为0的值，比如1

Swappiness

默认是5秒，太过频繁，可以适当地增加提交间隔

提交时间（Flush落盘时间）

操作系统参数

核心参数

Kafka消息组织结构实际为三级结构：主题-分区-消息，主题下的每条消息只会保存在某一个分区中，而不会在多个分区中被保存多份。

Kafka为何使用分区设计，而不直接使用多个主题？

为什么分区？

就是提供负载均衡的能力

为了实现系统的高伸缩性（ Scalability ）

实现业务级别的消息顺序的问题

分区的作用？

即顺序分配。比如一个主题下有 3 个分区，那么第一条消息被发送到分区0 ，第二条被发送到分区 1 ，第三条被发送到分区 2 ，以此类推。当生产第 4 条消息时又会重新开始，即将其分配到分区 0

轮询策略是 Kafka Java 生产者 API 默认提供的分区策略。如果你未指定 partitioner.class 参数，那么你的生产者程序会按照轮询的方式在主题的所有分区间均匀地 “ 码放 ” 消息。

总结：轮询策略有非常优秀的负载均衡表现，它总是能保证消息最大限度地被平均分配到所有分区上，故默认情况下它是最合理的分区策略，也是我们最常用的分区策略之一。

轮询策略（ Round-robin）

所谓随机就是我们随意地将消息放置到任意一个分区上

实现随机策略版的 partition 伪代码，只需两行代码即可：List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);return ThreadLocalRandom.current().nextInt(partitions.size());先计算出该主题总的分区数，然后随机地返回一个小于它的正整数。

本质上看随机策略也是力求将数据均匀地打散到各个分区，但从实际表现来看，它要逊于轮询策略，所以如果追求数据的均匀分布，还是使用轮询策略比较好。事实上，随机策略是老版本生产者使用的分区策略，在新版本中已经改为轮询了。

随机策略（Randomness）

特别是在 Kafka 不支持时间戳的年代，在一些场景中，工程师们都是直接将消息创建时间封装进 Key里面的。一旦消息被定义了 Key ，那么你就可以保证同一个 Key 的所有消息都进入到相同的分区里面，由于每个分区下的消息处理都是有顺序的，故这个策略被称为按消息键保序策略

Kafka 允许为每条消息定义消息键，简称为 Key 。这个 Key 的作用非常大，它可以是一个有着明确业务含义的字符串，比如客户代码、部门编号或是业务 ID 等；也可以用来表征消息元数据。

实现这个策略的 partition 方法同样简单，只需要下面两行代码即可：List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);return Math.abs(key.hashCode()) % partitions.size();

Kafka 默认分区策略实际上同时实现了两种策略：如果指定了 Key ，那么默认实现按消息键保序策略；如果没有指定 Key ，则使用轮询策略。

按消息键保序策略（ Key-ordering）

实例：根据 Broker 所在的 IP 地址实现定制化的分区策略List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);return partitions.stream().filter(p -> isSouth(p.leader().host())).map(PartitionInfo::partition).findAny().get();

这种策略一般只针对那些大规模的 Kafka 集群，特别是跨城市、跨国家甚至是跨大洲的集群。

基于地理位置的分区策略

其它分区策略

编写生产者程序时，你可以编写一个具体的类实现 org.apache.kafka.clients.producer.Partitioner 接口。这个接口也很简单，只定义了两个方法： partition() 和 close() ，通常你只需要实现最重要的 partition 方法。

这里的 topic 、 key 、 keyBytes 、 value 和 valueBytes 都属于消息数据， cluster 则是集群信息（比如当前 Kafka 集群共有多少主题、多少 Broker 等）。 Kafka 给你这么多信息，就是希望让你能够充分地利用这些信息对消息进行分区，计算出它要被发送到哪个分区中。只要你自己的实现类定义好了 partition 方法，同时设置 partitioner.class 参数为你自己实现类的 Full Qualified Name ，那么生产者程序就会按照你的代码逻辑对消息进行分区。虽说可以有无数种分区的可能，但比较常见的分区策略也就那么几种

需要显式地配置生产者端的参数 partitioner.class

自定义分区策略

分区策略？

生产者消息分区机制

不同的分区能够被放置到不同节点的机器上，而数据的读写操作也都是针对分区这个粒度而进行的，这样每个节点的机器都能独立地执行各自分区的读写请求处理。并且，我们还可以通过添加新的节点机器来增加整体系统的吞吐量。

V1版本

V2 版本是 Kafka 0.11.0.0 中正式引入的

V2版本

Kafka消息格式

（若干条）日志项（ record item ）

消息集合（ message set ）

消息（ message ）

Kafka消息层次

一个消息集合中包含若干条日志项（ record item ），而日志项才是真正封装消息的地方。 Kafka 底层的消息日志由一系列消息集合日志项组成。 Kafka 通常不会直接操作具体的一条条消息，它总是在消息集合这个层面上进行写入操作

比如在 Broker 端可能会对消息时间戳字段进行更新，那么重新计算之后的 CRC值也会相应更新

再比如 Broker 端在执行消息格式转换时（主要是为了兼容老版本客户端程序），也会带来 CRC 值的变化

原来在 V1 版本中，每条消息都需要执行 CRC 校验，但有些情况下消息的 CRC 值是会发生变化的

把消息的公共部分抽取出来放到外层消息集合里面，这样就不用每条消息都保存这些信息了

鉴于这些情况，再对每条消息都执行 CRC 校验就有点没必要了，不仅浪费空间还耽误 CPU 时间，因此在 V2 版本中，消息的 CRC 校验工作就被移到了消息集合这一层

之前 V1 版本中保存压缩消息的方法是把多条消息进行压缩然后保存到外层消息的消息体字段中

而 V2 版本的做法是对整个消息集合进行压缩。显然后者应该比前者有更好的压缩效果

保存压缩消息的方法发生了变化

为什么引入V2版本？

在相同条件下，不论是否启用压缩， V2 版本都比 V1 版本节省磁盘空间。当启用压缩时，这种节省空间的效果更加明显

怎么压缩

在 Kafka 中，压缩可能发生在两个地方：生产者端和 Broker 端

生产者程序中配置 compression.type 参数即表示启用指定类型的压缩算法

生产者端

大部分情况下 Broker 从 Producer 端接收到消息后仅仅是原封不动地保存而不会对其进行任何修改

实例：Producer端使用GZIP压缩，Broker端使用 Snappy 算法进行压缩？这种情况下 Broker 接收到 GZIP 压缩消息后，只能解压缩然后使用 Snappy 重新压缩一遍。如果你翻开 Kafka 官网，你会发现 Broker 端也有一个参数叫 compression.type ，和上面那个例子中的同名。但是这个参数的默认值是 producer ，这表示 Broker 端会 “ 尊重 ”Producer 端使用的压缩算法。可一旦你在 Broker 端设置了不同的 compression.type 值，就一定要小心了，因为可能会发生预料之外的压缩 / 解压缩操作，通常表现为 Broker 端 CPU 使用率飙升。

情况一： Broker 端指定了和 Producer 端不同的压缩算法

导致结果：Broker 端 CPU 使用率飙升

这个过程中会涉及消息的解压缩和重新压缩。一般情况下这种消息格式转换对性能是有很大影响的，除了这里的压缩之外，它还让Kafka 丧失了引以为豪的 Zero Copy 特性。

Zero Copy：说的是当数据在磁盘和网络进行传输时避免昂贵的内核态数据拷贝，从而实现快速的数据传输

所谓的消息格式转换主要是为了兼容老版本的消费者程序，在一个生产环境中， Kafka 集群中同时保存多种版本的消息格式非常常见。为了兼容老版本的格式， Broker 端会对新版本消息执行向老版本格式的转换。

情况二： Broker 端发生了消息格式转换

尽量保证消息格式的统一，这样不仅可以避免不必要的解压缩 / 重新压缩，对提升其他方面的性能也大有裨益

有两种例外情况就可能让 Broker 重新压缩消息

Broker 端

何时压缩

Consumer 端解压缩通常来说解压缩发生在消费者程序中，也就是说 Producer 发送压缩消息到Broker 后， Broker 照单全收并原样保存起来。当 Consumer 程序请求这部分消息时， Broker 依然原样发送出去，当消息到达 Consumer 端后，Consumer 自行解压缩还原成之前的消息

Kafka 会将启用了哪种压缩算法封装进消息集合中，这样当 Consumer 读取到消息集合时，它自然就知道了这些消息使用的是哪种压缩算法。

Consumer 怎么知道这些消息是用何种压缩算法压缩的呢？

每个压缩过的消息集合在 Broker 端写入时都要发生解压缩操作，目的就是为了对消息执行各种验证

Broker 端解压缩

总结： Producer 端压缩、 Broker 端保持、Consumer 端解压缩

何时解压缩

原先占 100 份空间的东西经压缩之后变成了占 20 份空间，那么压缩比就是 5 ，显然压缩比越高越好

压缩比

比如每秒能压缩或解压缩多少 MB 的数据。同样地，吞吐量也是越高越好

压缩/解压缩的吞吐量

衡量压缩算法优劣的两个重要指标

GZIP

Snappy

LZ4

Zstandard（简写为 zstd）

LZ4 > Snappy > zstd 和 GZIP

吞吐量

zstd > LZ4 > GZIP > Snappy

使用 Snappy 算法占用的网络带宽最多， zstd 最少，这是合理的，毕竟 zstd 就是要提供超高的压缩比

网络带宽

各个算法表现得差不多，只是在压缩时 Snappy 算法使用的 CPU 较多一些，而在解压缩时 GZIP 算法则可能使用更多的 CPU

CPU 使用率

Kafka中使用结论

各种压缩算法对比

Kafka 2.1.0 版本之前提供

从Kafka 2.1.0 开始正式支持

Producer 程序运行机器上的 CPU 资源要很充足。如果 Producer 运行机器本身 CPU 已经消耗殆尽了，那么启用消息压缩无疑是雪上加霜，只会适得其反

如果你的客户端机器 CPU 资源有很多富余，强烈建议你开启 zstd压缩，这样能极大地节省网络资源消耗

CPU资源充足

毕竟万兆网络还不是普通公司的标配，因此千兆网络中 Kafka 集群带宽资源耗尽这件事情就特别容易出现

带宽资源有限

何时启用压缩是比较合适的时机呢？

有条件的话尽量保证不要出现消息格式转换的情况

最佳实践

生产者压缩算法

Java生产者如何管理TCP连接

Java消费者如何管理TCP连接

Kafka副本机制

技术原理

生产者拦截器允许你在发送消息前以及消息提交成功后植入你的拦截器逻辑

生产者拦截器

消费者拦截器支持在消费消息前以及提交位移后编写特定逻辑

消费者拦截器

Kafka 拦截器

高级功能

这两种拦截器都支持链的方式，即你可以将一组拦截器串连成一个大的拦截器， Kafka 会按照添加顺序依次执行拦截器逻辑

集群监控

性能调优

Kafka 只对 “ 已提交 ” 的消息（ committed message ）做有限度的持久化保证

当 Kafka 的若干个 Broker 成功地接收到一条消息并写入到日志文件后，它们会告诉生产者程序这条消息已成功提交。此时，这条消息在 Kafka 看来就正式变为 “ 已提交 ” 消息了。

你可以选择只要有一个 Broker 成功保存该消息就算是已提交

也可以是令所有 Broker 都成功保存该消息才算是已提交

为什么是若干个 Broker 呢？这取决于你对 “ 已提交 ” 的定义

第一个核心要素是 “ 已提交的消息 ”

不论哪种情况， Kafka 只对已提交的消息做持久化保证这件事情是不变的。

这里的有限度就是说 Kafka 不丢消息是有前提条件的

假如你的消息保存在 N 个 Kafka Broker 上，那么这个前提条件就是这 N 个 Broker 中至少有 1个存活。只要这个条件成立， Kafka 就能保证你的这条消息永远不会丢失

第二个核心要素就是 “ 有限度的持久化保证 ”

Kafka 到底在什么情况下才能保证消息不丢失呢？

这种方式挺不靠谱，如果出现消息丢失，我们是无法知晓的。

目前 Kafka Producer 是异步发送消息的，也就是说如果你调用的是 producer.send(msg) 这个API ，那么它通常会立即返回，但此时你不能认为消息发送已成功完成。这种发送方式有个有趣的名字，叫 “fire and forget” （发射后不管），它的意思是，执行完一个操作后不去管它的结果是否成功。

网络抖动，导致消息压根就没有发送到 Broker 端

消息本身不合格导致 Broker 拒绝接收（比如消息太大了，超过了 Broker 的承受能力）

使用producer.send(msg) 导致消息没有发送成功的原因？

仅仅让 Producer 重试就可以了

瞬时错误

可以调整消息格式后再次发送

消息不合格

这里的callback可以针对消息提交失败的情况进行有针对性的处理，处理发送失败的责任在 Producer 端而非 Broker端。

fire and forget方式

场景：你写了一个 Producer 应用向 Kafka 发送消息，最后发现 Kafka 没有保存，于是大骂： “Kafka 真烂，消息发送居然都能丢失，而且还不告诉我？！

无论 Producer 端怎么重试都会失败的

Broker 都宕机

案例 1 ：生产者程序丢失数据

加入当前位移是90，我先将位移更新到100，然后开始消费消息，当消费到第95条时出错终止了消费，那么下次消费从100开始，96~99的消息就会丢失。

解决方案：维持先消费消息，再更新位移的顺序，就能最大限度地保证消息不丢失

可能带来的问题是消息的重复消费

将先消费消息，再更新位移顺序颠倒

Consumer 程序从 Kafka 获取到消息后开启了多个线程异步处理消息，而 Consumer 程序自动地向前更新位移。假如其中某个线程运行失败了，它负责的消息没有被成功处理，但位移已经被更新了，因此这条消息对于 Consumer 而言实际上是丢失了

解决方案：如果是多线程异步处理消费消息， Consumer 程序不要开启自动提交位移，而是要应用程序手动提交位移

多线程消费，某个线程运行失败导致

Consumer 端丢失数据主要体现在 Consumer 端要消费的消息不见了。 Consumer 程序有个 “ 位移 ” 的概念，表示的是这个 Consumer 当前消费到的 Topic 分区的位置。针对位移有2种情况会出现数据丢失

案例 2 ：消费者程序丢失数据

“ 消息丢失 ” 案例

设置 acks = all 。 acks 是 Producer 的一个参数，代表了你对 “ 已提交 ” 消息的定义。如果设置成all ，则表明所有副本 Broker 都要接收到消息，该消息才算是 “ 已提交 ” 。这是最高等级的 “ 已提交 ” 定义

设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失

设置 unclean.leader.election.enable = false 。这是 Broker 端的参数，它控制的是哪些 Broker有资格竞选分区的 Leader 。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的Leader ，必然会造成消息的丢失。故一般都要将该参数设置成 false ，即不允许这种情况的发生

设置 replication.factor >= 3 。这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余

设置 min.insync.replicas > 1 。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是 “ 已提交 ” 。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1

确保 replication.factor > min.insync.replicas 。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1

确保消息消费完成再提交。 Consumer 端有个参数 enable.auto.commit ，最好把它设置成false ，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的

无消息丢失配置实现

Kafka Streams

例如：kafka_2.11-2.2.1.tgz 前面的版本号是编译Kafka源代码的Scala编译器版本。所以这里实际Kafka的版本号为2.1.1。 Kafka服务端代码完全由Scala编写，而Kafka新版客户端代码完全由Java语言编写。（至于为什么不用Scala，只是因为社区来了一批Java程序员，老的Scala程序员隐退罢了）

由三部分组成，即“大版本号-小版本号-Patch号”

Kafka版本命名

开源的初始版本，只提供了最基础的消息队列功能，连副本机制都没有，不建议使用。

0.7

1、老版本客户端API存在很多问题，特别是生产者API，它默认使用同步方式发消息，吞吐量不会太高。虽然其支持异步方式，但实际场景可能会造成消息丢失，因此0.8.2.0版本社区引入了新版本Producer API，即需要指定Broker地址的Producer

2、对于使用0.8版本的建议至少升级到0.8.2.2这个版本，因为该版本中老版本消费者API是比较稳定的，这个版本不建议使用新版本Producer API，此时它的bug还非常多。

正式引入副本机制，成为真正意义上完备的分布式高可靠消息队列解决方案，需使用老版本客户端API，需指定Zookeeper的地址而非Broker地址

0.8

这是一个重量级的大版本更新，增加了基础的安全认证/权限功能

此时新版本Producer API在这个版本中算比较稳定了，线上环境可以切换新版Producer，但不要使用新版本Consumer API，因为Bug超多

同时使用Java重写了新版本消费者API

引入了Kafka Connect组件用于实现高性能的数据抽取