当前位置： 首页 > article >正文

消息中间件-kafka工作原理和配置

article 2025/1/13 16:21:05

kafka简介

kafka由 linked-in 开源搭建的解决信息生产和信息消费的消息系统的一个框架，它实现了生产者和消费者之间的无缝连接 ，具备高产出的分布式消息系统(A high-throughput distributed messaging system)。

Apache kafka 是一个分布式的基于push-subscribe的消息系统，和redis很类似，它具备快速、可扩展、可持久化的特点。，作为hadoop生态系统的一部分，最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/spark流式处理引擎。

Kafka的特性

  • 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒
  • 可扩展性：kafka集群支持热扩展
  • 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
  • 容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）
  • 高并发：支持数千个客户端同时读写

设计思想

  • Consumergroup：各个consumer可以组成一个组，每个消息只能被组中的一个consumer消费，如果一个消息可以被多个consumer消费的话，那么这些consumer必须在不同的组。
  • 消息状态：在Kafka中，消息的状态被保存在consumer中，broker不会关心哪个消息被消费了被谁消费了，只记录一个offset值（指向partition中下一个要被消费的消息位置），这就意味着如果consumer处理不好的话，broker上的一个消息可能会被消费多次。
  • 消息持久化：Kafka中会把消息持久化到本地文件系统中，并且保持极高的效率。
  • 消息有效期：Kafka会长久保留其中的消息，以便consumer可以多次消费，当然其中很多细节是可配置的。
  • 批量发送：Kafka支持以消息集合为单位进行批量发送，以提高push效率。
  • push-and-pull : Kafka中的Producer和consumer采用的是push-and-pull模式，即Producer只管向broker push消息，consumer只管从broker pull消息，两者对消息的生产和消费是异步的。
  • Kafka集群中broker之间的关系：不是主从关系，各个broker在集群中地位一样，可以随意的增加或删除任何一个broker节点。
  • 负载均衡方面： Kafka提供了一个 metadata API来管理broker之间的负载（对Kafka0.8.x而言，对于0.7.x主要靠zookeeper来实现负载均衡）。
  • 同步异步：Producer采用异步push方式，极大提高Kafka系统的吞吐率（可以通过参数控制是采用同步还是异步方式）。
  • 分区机制partition：Kafka的broker端支持消息分区，Producer可以决定把消息发到哪个分区，在一个分区中消息的顺序就是Producer发送消息的顺序，一个主题中可以有多个分区，具体分区的数量是可配置的。分区的意义很重大，后面的内容会逐渐体现。
  • 离线数据装载：Kafka由于对可拓展的数据持久化的支持，它也非常适合向Hadoop或者数据仓库中进行数据装载。
  • 插件支持：现在不少活跃的社区已经开发出不少插件来拓展Kafka的功能，如用来配合Storm、Hadoop、flume相关的插件。

应用场景

  • 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
  • 消息系统：解耦和生产者和消费者、缓存消息等。
  • 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
  • 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
  • 流式处理：比如spark streaming和storm
  • 事件源

架构组件

Kafka中发布订阅的对象是topic，可以为每类数据创建一个topic，把向topic发布消息的客户端称作producer，从topic订阅消息的客户端称作consumer。Producers和consumers可以同时从多个topic读写数据。一个kafka集群由一个或多个broker服务器组成，它负责持久化和备份具体的kafka消息。

  • topic：消息存放的目录即主题
  • Producer：生产消息到topic的一方
  • Consumer：订阅topic消费消息的一方
  • Broker：Kafka的服务实例就是一个broker

Kafka Topic&Partition

消息发送时都被发送到一个topic，其本质就是一个目录，而topic由是由一些Partition Logs(分区日志)组成,其组织结构如下图所示：

每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值。 
Kafka集群会保存所有的消息，不管消息有没有被消费；可以设定消息的过期时间，只有过期的数据才会被自动清除以释放磁盘空间。比如过期时间为2天，那么这2天内的所有消息都会被保存到集群中，数据只有超过了两天才会被清除。 
Kafka需要维持的元数据只有一个——消费消息在Partition中的offset值，Consumer每消费一个消息，offset就会加1。其实消息的状态完全是由Consumer控制的，Consumer可以跟踪和重设这个offset值，这样的话Consumer就可以读取任意位置的消息。 
把消息日志以Partition的形式存放有多重考虑，第一，方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；第二就是可以提高并发，因为可以以Partition为单位读写了。

涉及术语：

Producer：Producer即生产者，消息的产生者，是消息的入口。
kafka cluster：
        Broker：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1等……
        Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。
        Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！
        Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。
        Message：每一条发送的消息主体。
Consumer：消费者，即消息的消费方，是消息的出口。
Consumer Group：我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！
Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。

Kafka 核心组件

工作流程：

Replications、Partitions 和Leaders

kafka中的数据是持久化的并且能够容错的。Kafka允许用户为每个topic设置副本数量，副本数量决定了有几个broker来存放写入的数据。如果你的副本数量设置为3，那么一份数据就会被存放在3台不同的机器上，那么就允许有2个机器失败。一般推荐副本数量至少为2，这样就可以保证增减、重启机器时不会影响到数据消费。如果对数据持久化有更高的要求，可以把副本数量设置为3或者更多。 
Kafka中的topic是以partition的形式存放的，每一个topic都可以设置它的partition数量，Partition的数量决定了组成topic的log的数量。Producer在生产数据时，会按照一定规则（这个规则是可以自定义的）把消息发布到topic的各个partition中。上面的副本都是以partition为单位的，不过只有一个partition的副本会被选举成leader作为读写用。 
关于如何设置partition值需要考虑的因素。一个partition只能被一个消费者消费（一个消费者可以同时消费多个partition），因此，如果设置的partition的数量小于consumer的数量，就会有消费者消费不到数据。所以，推荐partition的数量一定要大于同时运行的consumer的数量。另外一方面，建议partition的数量大于集群broker的数量，这样leader partition就可以均匀的分布在各个broker中，最终使得集群负载均衡。在Cloudera,每个topic都有上百个partition。需要注意的是，kafka需要为每个partition分配一些内存来缓存消息数据，如果partition数量越大，就要为kafka分配更大的heap space。

Producers

Producers直接发送消息到broker上的leader partition，不需要经过任何中介一系列的路由转发。为了实现这个特性，kafka集群中的每个broker都可以响应producer的请求，并返回topic的一些元信息，这些元信息包括哪些机器是存活的，topic的leader partition都在哪，现阶段哪些leader partition是可以直接被访问的。 
Producer客户端自己控制着消息被推送到哪些partition。实现的方式可以是随机分配、实现一类随机负载均衡算法，或者指定一些分区算法。Kafka提供了接口供用户实现自定义的分区，用户可以为每个消息指定一个partitionKey，通过这个key来实现一些hash分区算法。比如，把userid作为partitionkey的话，相同userid的消息将会被推送到同一个分区。 
以Batch的方式推送数据可以极大的提高处理效率，kafka Producer 可以将消息在内存中累计到一定数量后作为一个batch发送请求。Batch的数量大小可以通过Producer的参数控制，参数值可以设置为累计的消息的数量（如500条）、累计的时间间隔（如100ms）或者累计的数据大小(64KB)。通过增加batch的大小，可以减少网络请求和磁盘IO的次数，当然具体参数设置需要在效率和时效性方面做一个权衡。 
Producers可以异步的并行的向kafka发送消息，但是通常producer在发送完消息之后会得到一个future响应，返回的是offset值或者发送过程中遇到的错误。这其中有个非常重要的参数“acks”,这个参数决定了producer要求leader partition 收到确认的副本个数，如果acks设置数量为0，表示producer不会等待broker的响应，所以，producer无法知道消息是否发送成功，这样有可能会导致数据丢失，但同时，acks值为0会得到最大的系统吞吐量。 
若acks设置为1，表示producer会在leader partition收到消息时得到broker的一个确认，这样会有更好的可靠性，因为客户端会等待直到broker确认收到消息。若设置为-1，producer会在所有备份的partition收到消息时得到broker的确认，这个设置可以得到最高的可靠性保证。 
Kafka 消息有一个定长的header和变长的字节数组组成。因为kafka消息支持字节数组，也就使得kafka可以支持任何用户自定义的序列号格式或者其它已有的格式如Apache Avro、protobuf等。Kafka没有限定单个消息的大小，但我们推荐消息大小不要超过1MB,通常一般消息大小都在1~10kB之前。

 Consumers

Kafka提供了两套consumer api，分为high-level api和sample-api。Sample-api 是一个底层的API，它维持了一个和单一broker的连接，并且这个API是完全无状态的，每次请求都需要指定offset值，因此，这套API也是最灵活的。 
在kafka中，当前读到消息的offset值是由consumer来维护的，因此，consumer可以自己决定如何读取kafka中的数据。比如，consumer可以通过重设offset值来重新消费已消费过的数据。不管有没有被消费，kafka会保存数据一段时间，这个时间周期是可配置的，只有到了过期时间，kafka才会删除这些数据。 
High-level API封装了对集群中一系列broker的访问，可以透明的消费一个topic。它自己维持了已消费消息的状态，即每次消费的都是下一个消息。 
High-level API还支持以组的形式消费topic，如果consumers有同一个组名，那么kafka就相当于一个队列消息服务，而各个consumer均衡的消费相应partition中的数据。若consumers有不同的组名，那么此时kafka就相当与一个广播服务，会把topic中的所有消息广播到每个consumer。

consumer和partition对应关系

注意：1个partition只能被同组的一个consumer消费，同组的consumer则起到均衡效果

1、消费者多于partition

创建一个topic——test （topic： test 只有一个partition）

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

在g2组中启动两个consumer：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config config/consumer_g2.properties
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config config/consumer_g2.properties

消费者数量为2大于partition数量1，此时partition和消费者进程对应关系为：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g2
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test 0 9 9 0 consumer-1-4a2a4aa8-32f4-4904-9c16-1c0bdf7128a2 /127.0.0.1 consumer-1
- - - - - consumer-1-fd7b120f-fd21-4e07-8c23-87b71c1ee8a5 /127.0.0.1 consumer-1

消费者consumer-1-fd7b120f-fd21-4e07-8c23-87b71c1ee8a5无对应的partition：

存在消费者无法消费到数据的情况。向test发送消息：1，2， 3，4，5，6，7，8，9
只有C1能接收到消息，C2则不能接收到消息，即同一个partition内的消息只能被同一个组中的一个consumer消费。当消费者数量多于partition的数量时，多余的消费者空闲。
也就是说如果只有一个partition，你在同一组启动多少个consumer都没用，partition的数量决定了此topic在同一组中被可被均衡的程度，例如partition=4，则可在同一组中被最多4个consumer均衡消费。

2、消费者少于和等于partition

topic：test2包含3个partition

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic test2

在g3组中启动2个consumer：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test2 --from-beginning --consumer.config config/consumer_g3.properties
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test2 --from-beginning --consumer.config config/consumer_g3.properties

消费者数量为2小于partition数量3，此时partition和消费者进程对应关系为： 

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test2 0 8 8 0 consumer-1-8b872ef7-a2f0-4bd3-b2a8-7b26e4d8ab2c /127.0.0.1 consumer-1
test2 1 7 7 0 consumer-1-8b872ef7-a2f0-4bd3-b2a8-7b26e4d8ab2c /127.0.0.1 consumer-1
test2 2 8 8 0 consumer-1-f362847d-1094-4895-ad8b-1e1f1c88936c /127.0.0.1 consumer-1

其中，consumer-1-8b872ef7-a2f0-4bd3-b2a8-7b26e4d8ab2c对应了2个partition：

向test2发送消息1，2，3，4，5，6，7，8，9
C1接收到1，3，4，6，7，9
C2接收到2，5，8
此时P1、P2对对应C1，即多个partition对应一个消费者，C1接收到消息量是C2的两倍

再在g3组中再启动一个消费者，使得消费者数量为3等于topic2中partition的数量：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test2 --from-beginning --consumer.config config/consumer_g3.properties

对应关系如下：

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test2 0 8 8 0 consumer-1-8b872ef7-a2f0-4bd3-b2a8-7b26e4d8ab2c /127.0.0.1 consumer-1
test2 1 7 7 0 consumer-1-ab472ed5-de11-4e56-863a-67bf3a3cc36a /127.0.0.1 consumer-1
test2 2 8 8 0 consumer-1-f362847d-1094-4895-ad8b-1e1f1c88936c /127.0.0.1 consumer-1

向test2发送消息1，2，3，4，5，6，7，8，9
C1接收到了：2，5，8
C2接收到了：3，6，9
C3接收到了：1，4，7
C1，C2，C3均分了test2的所有消息，即消息在同一个组之间的消费者之间均分!

3、多个消费者组

启动g4组，仅包含一个消费者C1，消费topic2的消息，此时消费端有两个消费者组：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test2 --from-beginning --consumer.config config/consumer_g4.properties --delete-consumer-offsets

g4组的C1的对应了test2的所有partition:

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g4
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test2 0 36 36 0 consumer-1-befc9234-260d-4ad3-b283-b67a2bf446ca /127.0.0.1 consumer-1
test2 1 35 35 0 consumer-1-befc9234-260d-4ad3-b283-b67a2bf446ca /127.0.0.1 consumer-1
test2 2 36 36 0 consumer-1-befc9234-260d-4ad3-b283-b67a2bf446ca /127.0.0.1 consumer-1

向test2发送消息1，2，3，4，5，6，7，8，9，消息被g3组的消费者均分，g4组的消费者在接收到了所有的消息。
g3组：
C1接收到了：2，5，8
C2接收到了：3，6，9
C3接收到了：1，4，7
g4组：
C1接收到了：1，2，3，4，5，6，7，8，9
启动多个组，则会使同一个消息被消费多次。

具体地：

Consumer Rebalance的算法如下：

  • 将目标Topic下的所有Partirtion排序，存于PT
  • 对某Consumer Group下所有Consumer排序，存于CG，第ii个Consumer记为Ci
  • N=size(PT)/size(CG)，向上取整
  • 解除Ci对原来分配的Partition的消费权（i从0开始）
  • 将第i∗N到（i+1）∗N−1个Partition分配给Ci

Kafka核心特性

压缩

Kafka支持以集合（batch）为单位发送消息，在此基础上，Kafka还支持对消息集合进行压缩，Producer端可以通过GZIP或Snappy格式对消息集合进行压缩。Producer端进行压缩之后，在Consumer端需进行解压。压缩的好处就是减少传输的数据量，减轻对网络传输的压力，在对大数据处理上，瓶颈往往体现在网络上而不是CPU（压缩和解压会耗掉部分CPU资源）。 
那么如何区分消息是压缩的还是未压缩的呢，Kafka在消息头部添加了一个描述压缩属性字节，这个字节的后两位表示消息的压缩采用的编码，如果后两位为0，则表示消息未被压缩。

消息可靠性

在消息系统中，保证消息在生产和消费过程中的可靠性是十分重要的，在实际消息传递过程中，可能会出现如下三中情况：

  • 一个消息发送失败
  • 一个消息被发送多次
  • 最理想的情况：exactly-once ,一个消息发送成功且仅发送了一次

有许多系统声称它们实现了exactly-once，但是它们其实忽略了生产者或消费者在生产和消费过程中有可能失败的情况。比如虽然一个Producer成功发送一个消息，但是消息在发送途中丢失，或者成功发送到broker，也被consumer成功取走，但是这个consumer在处理取过来的消息时失败了。 
从Producer端看：Kafka是这么处理的，当一个消息被发送后，Producer会等待broker成功接收到消息的反馈（可通过参数控制等待时间），如果消息在途中丢失或是其中一个broker挂掉，Producer会重新发送（Kafka有备份机制，可以通过参数控制是否等待所有备份节点都收到消息）。 
从Consumer端看：前面讲到过partition，broker端记录了partition中的一个offset值，这个值指向Consumer下一个即将消费message。当Consumer收到了消息，但却在处理过程中挂掉，此时Consumer可以通过这个offset值重新找到上一个消息再进行处理。Consumer还有权限控制这个offset值，对持久化到broker端的消息做任意处理。

 备份机制

备份机制是Kafka0.8版本的新特性，备份机制的出现大大提高了Kafka集群的可靠性、稳定性。有了备份机制后，Kafka允许集群中的节点挂掉后而不影响整个集群工作。一个备份数量为n的集群允许n-1个节点失败。在所有备份节点中，有一个节点作为lead节点，这个节点保存了其它备份节点列表，并维持各个备份间的状体同步:

Kafka高效性相关设计

消息的持久化

Kafka高度依赖文件系统来存储和缓存消息，一般的人认为磁盘是缓慢的，这导致人们对持久化结构具有竞争性持怀疑态度。其实，磁盘远比你想象的要快或者慢，这决定于我们如何使用磁盘。 
一个和磁盘性能有关的关键事实是：线性写的速度远远大于随机写。比如：在一个6 7200rpm SATA RAID-5 的磁盘阵列上线性写的速度大概是600M/秒，但是随机写的速度只有100K/秒，两者相差将近6000倍。线性读写在大多数应用场景下是可以预测的，因此，操作系统利用read-ahead和write-behind技术来从大的数据块中预取数据，或者将多个逻辑上的写操作组合成一个大写物理写操作中。为了补偿这个性能上的分歧，现代操作系统都会把空闲的内存用作磁盘缓存，尽管在内存回收的时候会有一点性能上的代价。所有的磁盘读写操作会在这个统一的缓存上进行。 
此外，如果在JVM的基础上构建kafka应该知道：

  1. 一个对象的内存消耗是非常高的，经常是所存数据的两倍或者更多。
  2. 随着堆内数据的增多，Java的垃圾回收会变得非常昂贵。

基于这些事实，利用文件系统并且依靠页缓存比维护一个内存缓存或者其他结构要好——至少要使得可用的缓存加倍，通过自动访问可用内存，并且通过存储更紧凑的字节结构而不是一个对象，这将有可能再次加倍。这么做的结果就是在一台32GB的机器上，如果不考虑GC惩罚，将最多有28-30GB的缓存。此外，这些缓存将会一直存在即使服务重启，然而进程内缓存需要在内存中重构（10GB缓存需要花费10分钟）或者它需要一个完全冷缓存启动（非常差的初始化性能）。它同时也简化了代码，因为现在所有的维护缓存和文件系统之间内聚的逻辑都在操作系统内部了，这使得这样做比one-off in-process attempts更加高效与准确。如果你的磁盘应用更加倾向于顺序读取，那么read-ahead在每次磁盘读取中实际上获取到这人缓存中的有用数据。 
以上这些建议了一个简单的设计：不同于维护尽可能多的内存缓存并且在需要的时候刷新到文件系统中，换一种思路。所有的数据不需要调用刷新程序，而是立刻将它写到一个持久化的日志中。事实上，这仅仅意味着，数据将被传输到内核页缓存中并稍后被刷新。可以增加一个配置项以让系统的用户来控制数据在什么时候被刷新到物理硬盘上。

常数时间性能保证

消息系统中持久化数据结构的设计通常是维护一个和消费队列有关的B树或者其它能够随机存取结构的元数据信息。B树是一个很好的结构，可以用在事务型与非事务型的语义中。但是它需要一个很高的花费，尽管B树的操作需要O(logN)。通常情况下，这被认为与常数时间等价，但这对磁盘操作来说是不对的。磁盘寻道一次需要10ms，并且一次只能寻一个，因此并行化是受限的。 
直觉上来讲，一个持久化的队列可以构建在对一个文件的读和追加上，就像一般情况下的日志解决方案。尽管和B树相比，这种结构不能支持丰富的语义，但是它有一个优点，所有的操作都是常数时间，并且读写之间不会相互阻塞。这种设计具有极大的性能优势：最终系统性能和数据大小完全无关，服务器可以充分利用廉价的硬盘来提供高效的消息服务。 
事实上还有一点，磁盘空间的无限增大而不影响性能这点，意味着可以提供一般消息系统无法提供的特性。比如说，消息被消费后不是立马被删除，可以将这些消息保留一段相对比较长的时间（比如一个星期）。

进一步提高效率

有一种非常主要的应用场景是：处理Web活动数据，它的特点是数据量非常大，每一次的网页浏览都会产生大量的写操作。更进一步，假设每一个被发布的消息都会被至少一个consumer消费，因此需要让消费变得更廉价。 如以下应用场景：

  • 太多小的I/O操作
  • 过多的字节拷贝

为了减少大量小I/O操作的问题，kafka的协议是围绕消息集合构建的。Producer一次网络请求可以发送一个消息集合，而不是每一次只发一条消息。在server端是以消息块的形式追加消息到log中的，consumer在查询的时候也是一次查询大量的线性数据块。消息集合即MessageSet，实现本身是一个非常简单的API，它将一个字节数组或者文件进行打包。所以对消息的处理，这里没有分开的序列化和反序列化的上步骤，消息的字段可以按需反序列化（如果没有需要，可以不用反序列化）。 
另一个影响效率的问题就是字节拷贝。为了解决字节拷贝的问题，kafka设计了一种“标准字节消息”，Producer、Broker、Consumer共享这一种消息格式。Kakfa的message log在broker端就是一些目录文件，这些日志文件都是MessageSet按照这种“标准字节消息”格式写入到磁盘的。 
维持这种通用的格式对这些操作的优化尤为重要：持久化log 块的网络传输。流行的unix操作系统提供了一种非常高效的途径来实现页面缓存和socket之间的数据传递。在Linux操作系统中，这种方式被称作：sendfile system call（Java提供了访问这个系统调用的方法：FileChannel.transferTo api）。

为了理解sendfile的影响，需要理解一般的将数据从文件传到socket的路径：

  1. 操作系统将数据从磁盘读到内核空间的页缓存中
  2. 应用将数据从内核空间读到用户空间的缓存中
  3. 应用将数据写回内核空间的socket缓存中
  4. 操作系统将数据从socket缓存写到网卡缓存中，以便将数据经网络发出

这种操作方式明显是非常低效的，这里有四次拷贝，两次系统调用。如果使用sendfile，就可以避免两次拷贝：操作系统将数据直接从页缓存发送到网络上。所以在这个优化的路径中，只有最后一步将数据拷贝到网卡缓存中是需要的。 但是最终期望一个主题上有多个消费者是一种常见的应用场景。利用上述的zero-copy，数据只被拷贝到页缓存一次，然后就可以在每次消费时被重得利用，而不需要将数据存在内存中，然后在每次读的时候拷贝到内核空间中。这使得消息消费速度可以达到网络连接的速度。这样以来，通过页面缓存和sendfile的结合使用，整个kafka集群几乎都已以缓存的方式提供服务，而且即使下游的consumer很多，也不会对整个集群服务造成压力。 

Kafka集群部署

集群部署

为了提高性能，推荐采用专用的服务器来部署kafka集群，尽量与hadoop集群分开，因为kafka依赖磁盘读写和大的页面缓存，如果和hadoop共享节点的话会影响其使用页面缓存的性能。 
Kafka集群的大小需要根据硬件的配置、生产者消费者的并发数量、数据的副本个数、数据的保存时长综合确定。 
磁盘的吞吐量尤为重要，因为通常kafka的瓶颈就在磁盘上。 
Kafka依赖于zookeeper，建议采用专用服务器来部署zookeeper集群，zookeeper集群的节点采用偶数个，一般建议用3、5、7个。注意zookeeper集群越大其读写性能越慢，因为zookeeper需要在节点之间同步数据。一个3节点的zookeeper集群允许1个节点失败，一个5节点集群允许2个节点失败。

集群大小

有很多因素决定着kafka集群需要具备存储能力的大小，最准确的衡量办法就是模拟负载来测算一下，Kafka本身也提供了负载测试的工具。 
如果不想通过模拟实验来评估集群大小，最好的办法就是根据硬盘的空间需求来推算。根据网络和磁盘吞吐量需求来做一下估算。 
做如下假设：

  • W：每秒写多少MB
  • R ：副本数
  • C ：Consumer的数量

一般的来说，kafka集群瓶颈在于网络和磁盘吞吐量，所以先评估一下集群的网络和磁盘需求。 对于每条消息，每个副本都要写一遍，所以整体写的速度是W*R。读数据的部分主要是集群内部各个副本从leader同步消息读和集群外部的consumer读，所以集群内部读的速率是(R-1)*W,同时，外部consumer读的速度是C*W，因此：

  • Write：W*R
  • Read：（R-1）*W+C*W

需要注意在读的时候缓存部分数据来减少IO操作，如果一个集群有M MB内存，写的速度是W MB/sec，则允许M/(W*R) 秒的写可以被缓存。如果集群有32GB内存，写的速度是50MB/s的话，则可以至少缓存10分钟的数据。

Kafka性能测试

Performance testing

Kafka在zookeeper中的数据结构

Kafka data structures in Zookeeper

kafka配置

#配置java，https://www.oracle.com/java/technologies/downloads/archive/
#下载Java SE 8 (8u202)jdk-8u202-windows-x64.exe
#下载完后安装在不带空格的路径下，不要安装在program file下
#安装好后配置环境变量：
#新建变量名：JAVA_HOME 变量值：C:\Users\59980\java\bin
#新建变量名：CLASSPATH 变量值：.;%JAVA_HOME%\lib\dt.jar;%JAVA_HOME%\lib\tools.jar
#在path路径下append：%JAVA_HOME%\bin;%JAVA_HOME%\jre\bin
#测试是否配置好输入：java -version 显示java版本即完成配置
#下载kafka：https://mirrors.aliyun.com/apache/kafka/ 版本：kafka_2.13-3.7.2.tgz
#或https://pan.baidu.com/s/1qD06L8_OLbe7NFmQI3Ja6g?from=init&pwd=2024
#Kafka2.2.0开始支持使用内置的ZooKeeper替代外部ZooKeeper，所以不需要安装Zookeeper,直接解压即可
#下载好后直接解压在C盘下面，目录放的不要过深，如：C:\kafka_2.13-3.7.2

因为Kafka中的Broker注册，Topic注册，以及负载均衡都是在Zookeeper中管理，所以需要先启动内置的Zookeeper，打开Config文件夹下的zookeeper.properties文件，修改dataDir目录路径： 

dataDir=C:/kafka_2.13-3.7.2/zookeeperData # 本身是没有该目录的，是新建

修改Config文件夹下的server.properties文件，修改kafka的日志文件路径：

log.dirs=C:/kafka_2.13-3.7.2/kafka-logs

同时开启远程（本地的localhost_ip）：

启动Zookeeper：

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

当看到绑定到IP地址为0.0.0.0、端口号为2181的地址，表示ZooKeeper服务器监听在该地址，启动成功： 

新建一个cmd窗口， 启动kafka：

.\bin\windows\kafka-server-start.bat .\config\server.properties

监听窗口启动成功： 

kafka自启脚本：

# kafka_procedure.bat

@echo off
 
REM 设置绝对路径
set KAFKA_HOME=C:\kafka_2.13-3.7.2
set ZOOKEEPER_CONFIG=%KAFKA_HOME%\config\zookeeper.properties
set KAFKA_CONFIG=%KAFKA_HOME%\config\server.properties
 
REM 删除 kafka_log 目录及其内容
if exist %KAFKA_HOME%\kafka_log (
    echo 删除 kafka_log 目录...
    rmdir /s /q %KAFKA_HOME%\kafka_log
)
 
REM 删除 zookeeperData 目录
if exist %KAFKA_HOME%\bin\windows\kafka_2.13-3.7.2zookeeperData (
    echo 删除 Kafkakafka_2.13-3.5.0zookeeperData 目录...
    rmdir /s /q %KAFKA_HOME%\bin\windows\kafka_2.13-3.7.2zookeeperData
)
 
REM 启动Zookeeper
cd /d %KAFKA_HOME%\bin\windows
start /b zookeeper-server-start.bat %ZOOKEEPER_CONFIG%
 
REM 等待Zookeeper启动
timeout /t 10 /nobreak
 
REM 检查 meta.properties 文件
if exist %KAFKA_HOME%\kafka_log\meta.properties (
    echo "Meta properties already exist. Starting Kafka server..."
) else (
    echo "Meta properties not found. Starting Kafka server for the first time..."
)
 
REM 启动Kafka服务器
start /b kafka-server-start.bat %KAFKA_CONFIG%
 
exit

将脚本后缀改为bat，放到windwos自启中：

Kafka定位为分布式消息发布-订阅系统，提及分布式就可以想象只有当在多节点、高并发环境下才能最大的发挥它的价值 。

python-kafka生产者消费者demo

生产者，自动创建test主题，并每隔一秒发送一条数据： 

# python3.8版本安装kafka-python不支持了，建议用3.9以上安装：pip install kafka-python
from kafka import KafkaProducer
import json
import time
 
# Kafka服务器地址
bootstrap_servers = ['192.168.71.24:9092']
 
# 创建KafkaProducer实例
producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
 
# 发送消息的函数
def send_message(topic, message):
    # 将消息转换为字节
    producer.send(topic, json.dumps(message).encode('utf-8'))
    producer.flush()
 
if __name__ == '__main__':
    # 创建'test'主题
    topic = 'test'
    # 发送消息
    i = 1
    while True:
        message = {'num': i, 'msg': f'Hello Kafka {i}'}
        send_message(topic, message)
        i += 1
        time.sleep(1)

消费者：

from kafka import KafkaConsumer
import json
 
# Kafka服务器地址
bootstrap_servers = ['192.168.71.24:9092']
 
# 创建KafkaConsumer实例
consumer = KafkaConsumer(
    'test',
    bootstrap_servers=bootstrap_servers,
    auto_offset_reset='latest',  # 从最新的消息开始消费
    # auto_offset_reset='earliest',  # 从最早的offset开始消费
    enable_auto_commit=True,  # 自动提交offset
    group_id='my-group'  # 消费者组ID
)
 
# 消费消息
for message in consumer:
    # 将接收到的消息解码并转换为字典
    message = json.loads(message.value.decode('utf-8'))
    print(f"Received message: {message}")

消费者参数如下：

1、auto_offset_reset
该参数指定了当Kafka中没有初始偏移量或当前偏移量在服务器上不再存在时（例如数据被删除了），消费者应从何处开始读取数据。
可选值：
earliest：从最早的记录开始消费，即从分区日志的开始处开始。
latest：从最新的记录开始消费，即从分区日志的末尾开始。（默认）
none：如果没有为消费者指定初始偏移量，就抛出一个异常。

2、enable_auto_commit

该参数指定了消费者是否周期性地提交它所消费的偏移量。自动提交偏移量可以简化消费者的使用，但可能有重复消费或数据丢失的风险。禁用自动提交可以更精确地控制偏移量的提交时机，通常在确保消息处理成功后才提交偏移量。
可选值：
true：自动提交偏移量。（默认）
false：不自动提交偏移量，需要手动调用commitSync()或commitAsync()来提交偏移量。

3、group_id

该参数用于指定消费者所属的消费组。同一个消费组的消费者将共同消费一个主题的不同分区，而不同消费组的消费者可以独立地消费消息，互不影响。这对于实现负载均衡和故障转移很有用。
类型：字符串（必须指定）

list_topics()：获取主题元数据。

create_topics()：创建新主题。

delete_topics()：删除主题。

from kafka.admin import KafkaAdminClient, NewTopic
 
# 获取主题元数据
admin_client = KafkaAdminClient(bootstrap_servers='ip:9092', client_id='test')
topics = admin_client.list_topics()
print(topics)
 
 
# 创建主题
new_topic = NewTopic(name="test-topic", num_partitions=3, replication_factor=1)
admin_client.create_topics(new_topics=[new_topic], validate_only=False)
 
# 删除主题
admin_client.delete_topics(topics=['test-topic'])

结果如下，在多节点情况下同样设置即可：


http://www.kler.cn/a/500977.html

相关文章：

  • HTTP-响应协议
  • webpack打包要义
  • 通信与网络安全之网络连接
  • 微信小程序——创建滑动颜色条
  • Browser-Use Web UI：浏览器自动化与AI的完美结合
  • 依据正则表达式拦截文本
  • FreePBX 17 on ubuntu24 with Asterisk 20
  • 【Spring】@Size 无法拦截null的原因
  • Julia语言的软件工程
  • .NET Core NPOI 导出图片到Excel指定单元格并自适应宽度
  • Cesium加载地形
  • STM32 : 波特率发生器
  • ceph 数据均衡
  • K11082 加密与解密
  • Vue2+OpenLayers使用Overlay实现点击获取当前经纬度信息（提供Gitee源码）
  • 数据结构讲解
  • es分页查询
  • （STM32笔记）十二、DMA的基础知识与用法 第二部分
  • el-table横向滚动条，滚动后消失
  • 解锁动态规划的奥秘：从零到精通的创新思维解析（5）
  • MathBuddyGUI：带控制系统仿真功能、积分运算的计算器，MATLAB课程设计
  • Golang笔记——Interface类型
  • Audio音频输出通道
  • defineProps 和 Pinia 的使用方法和区别总结
  • Mac Mini 最优雅的备份方式：使用极空间实现自动整机备份
  • 基于ILI9341液晶屏+STM32U5单片的显示试验