自建网站h5农安建设局网站

kafka 如何确保消息的可靠性传输

消费端弄丢了数据

唯一可能导致消费者弄丢数据的情况，就是消费到了这个消息，然后还没处理就自动提交了offset，让kafka以为你已经消费好了这个消息。

对于消费端来说只要关闭自动提交offset，在处理完之后自己手动提交offset，就可以保证数据不会丢。但是此时确实还是会重复消费，比如你刚处理完，还没提交offset，结果自己挂了，此时肯定会重复消费一次，自己保证幂等性就好了。

kafka弄丢了数据

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

一般要求设置如下4个参数：

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

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

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

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

生产者会不会弄丢数据

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

Kafka 高性能的体现

利用Partition实现并行处理

Kafka中每个Topic都包含一个或多个Partition，不同Partition可位于不同节点。同时Partition在物理上对应一个本地文件夹，每个Partition包含一个或多个Segment，每个Segment包含一个数据文件和一个与之对应的索引文件。在逻辑上，可以把一个Partition当作一个非常长的数组，可通过这个“数组”的索引（offset）去访问其数据。

一方面，由于不同Partition可位于不同机器，因此可以充分利用集群优势，实现机器间的并行处理。另一方面，由于Partition在物理上对应一个文件夹，即使多个Partition位于同一个节点，也可通过配置让同一节点上的不同Partition置于不同的disk drive上，从而实现磁盘间的并行处理，充分发挥多磁盘的优势。

利用多磁盘的具体方法是，将不同磁盘mount到不同目录，然后在server.properties中，将log.dirs设置为多目录（用逗号分隔）。Kafka会自动将所有Partition尽可能均匀分配到不同目录也即不同目录（也即不同disk）上。

Partition是最小并发粒度，Partition个数决定了可能的最大并行度。

利用PageCache

Page Cache，又称pcache，其中文名称为页高速缓冲存储器，简称页高缓。page cache的大小为一页，通常为4K。在linux读写文件时，它用于缓存文件的逻辑内容，从而加快对磁盘上映像和数据的访问。 是Linux操作系统的一个特色。

读Cache

当内核发起一个读请求时(例如进程发起read()请求)，首先会检查请求的数据是否缓存到了Page Cache中。

如果有，那么直接从内存中读取，不需要访问磁盘，这被称为cache命中(cache hit)；

如果cache中没有请求的数据，即cache未命中(cache miss)，就必须从磁盘中读取数据。然后内核将读取的数据缓存到cache中，这样后续的读请求就可以命中cache了。

page可以只缓存一个文件部分的内容，不需要把整个文件都缓存进来。

写Cache

当内核发起一个写请求时(例如进程发起write()请求)，同样是直接往cache中写入，后备存储中的内容不会直接更新(当服务器出现断电关机时，存在数据丢失风险)。

内核会将被写入的page标记为dirty，并将其加入dirty list中。内核会周期性地将dirty list中的page写回到磁盘上，从而使磁盘上的数据和内存中缓存的数据一致。

当满足以下两个条件之一将触发脏数据刷新到磁盘操作：

• 数据存在的时间超过了dirty_expire_centisecs(默认300厘秒，即30秒)时间；
• 脏数据所占内存 > dirty_background_ratio，也就是说当脏数据所占用的内存占总内存的比例超过dirty_background_ratio(默认10，即系统内存的10%)的时候会触发pdflush刷新脏数据。

如何查看Page Cache参数

执行命令 sysctl -a|grep dirty

如何提高 Kafka 性能

调整内核参数来优化IO性能

1.vm.dirty_background_ratio参数优化

这个参数指定了当文件系统缓存脏页数量达到系统内存百分之多少时（如5%）就会触发后台回写进程运行，将一定缓存的脏页异步地刷入磁盘；

当cached中缓存当数据占总内存的比例达到这个参数设定的值时将触发刷磁盘操作。

把这个参数适当调小，这样可以把原来一个大的IO刷盘操作变为多个小的IO刷盘操作，从而把IO写峰值削平。对于内存很大和磁盘性能比较差的服务器，应该把这个值设置的小一点。

2.vm.dirty_ratio参数优化

这个参数则指定了当文件系统缓存脏页数量达到系统内存百分之多少时（如10%），系统不得不开始处理缓存脏页（因为此时脏页数量已经比较多，为了避免数据丢失需要将一定脏页刷入外存）；在此过程中很多应用进程可能会因为系统转而处理文件IO而阻塞。

对于写压力特别大的，建议把这个参数适当调大；对于写压力小的可以适当调小；如果cached的数据所占比例(这里是占总内存的比例)超过这个设置，

系统会停止所有的应用层的IO写操作，等待刷完数据后恢复IO。所以万一触发了系统的这个操作，对于用户来说影响非常大的。

3.vm.dirty_expire_centisecs参数优化

这个参数会和参数vm.dirty_background_ratio一起来作用，一个表示大小比例，一个表示时间；即满足其中任何一个的条件都达到刷盘的条件。

为什么要这么设计呢？如果只有参数 vm.dirty_background_ratio ，也就是说cache中的数据需要超过这个阀值才会满足刷磁盘的条件；如果数据一直没有达到这个阀值，那相当于cache中的数据就永远无法持久化到磁盘，这种情况下，一旦服务器重启，那么cache中的数据必然丢失。

结合以上情况，所以添加了一个数据过期时间参数。当数据量没有达到阀值，但是达到了我们设定的过期时间，同样可以实现数据刷盘。

4.vm.dirty_writeback_centisecs参数优化

理论上调小这个参数，可以提高刷磁盘的频率，从而尽快把脏数据刷新到磁盘上。但一定要保证间隔时间内一定可以让数据刷盘完成。

5.vm.swappiness参数优化

禁用swap空间，设置vm.swappiness=0

减少网络开销批处理

批处理是一种常用的用于提高I/O性能的方式。对Kafka而言，批处理既减少了网络传输的Overhead，又提高了写磁盘的效率。

Kafka 的send方法并非立即将消息发送出去，而是通过batch.size和linger.ms控制实际发送频率，从而实现批量发送。

由于每次网络传输，除了传输消息本身以外，还要传输非常多的网络协议本身的一些内容（称为Overhead），所以将多条消息合并到一起传输，可有效减少网络传输的Overhead，进而提高了传输效率。

数据压缩降低网络负载

Kafka支持将数据压缩后再传输给Broker。除了可以将每条消息单独压缩然后传输外，Kafka还支持在批量发送时，将整个Batch的消息一起压缩后传输。数据压缩的一个基本原理是，重复数据越多压缩效果越好。因此将整个Batch的数据一起压缩能更大幅度减小数据量，从而更大程度提高网络传输效率。

Broker接收消息后，并不直接解压缩，而是直接将消息以压缩后的形式持久化到磁盘。Consumer Fetch到数据后再解压缩。因此Kafka的压缩不仅减少了Producer到Broker的网络传输负载，同时也降低了Broker磁盘操作的负载，也降低了Consumer与Broker间的网络传输量，从而极大得提高了传输效率，提高了吞吐量。

高效的序列化方式

Kafka消息的Key和Value的类型可自定义，只需同时提供相应的序列化器和反序列化器即可。

因此用户可以通过使用快速且紧凑的序列化-反序列化方式（如Avro，Protocal Buffer）来减少实际网络传输和磁盘存储的数据规模，从而提高吞吐率。这里要注意，如果使用的序列化方法太慢，即使压缩比非常高，最终的效率也不一定高。