Kafka

2025-03-26 23:22:09   0  举报





AI智能生成

Apache Kafka是一个开源流处理平台，设计用于构建可扩展的实时数据管道和流应用程序。它能够在各种不同系统间高效地移动数据，并支持数据的发布和订阅模式。Kafka的核心内容包括其分布式架构，容错性，高吞吐量，以及水平扩展的能力。文件类型方面，Kafka主要使用配置文件和日志文件。其配置文件是用于调整服务器运行参数和客户端行为的关键文件，通常包含如`server.properties`等。日志文件则保存了主题中的消息数据，它们以分区的形式存储在服务器上。Kafka的修饰语可能涉及其高性能、易用性、可扩展性、以及在大数据处理场景中的广泛应用。整体而言，Kafka作为一个强大的消息队列系统，广泛应用于日志聚合、事件源架构和流式处理等需要高效、可靠数据传递机制的场景中。

消息队列

kafka

作者其他创作

大纲/内容

基础

介绍

简介：Kafka是一个分布式的，可划分的，冗余备份的持久性的日志服务。它主要用于处理活跃的流式数据。

消息系统术语：消息归纳topic，发布消息producers，消费消息consumer，集群服务broker。

关系图

图片链接

总结：两个作用，1.降低系统组网复杂度。2.降低编程复杂度。

Kafka独特设计在什么地方？

1.同时为发布和订阅提供高吞吐量。据了解，Kafka每秒可以生产约25万消息（50 MB），每秒处理55万消息（110 MB）。

2.可进行持久化操作。将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。通过将数据持久化到硬盘以及replication防止数据丢失。

3.分布式系统，易于向外扩展。所有的producer、broker和consumer都会有多个，均为分布式的。无需停机即可扩展机器。

4.消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡。

5.支持online和offline的场景。

kafka架构

架构

producer，broker，consumer关系图

Kafka的整体架构非常简单，是显式分布式架构，producer、broker（kafka）和consumer都可以有多个。

Producer，consumer实现Kafka注册的接口，数据从producer发送到broker，broker承担一个中间缓存和分发的作用。

broker分发注册到系统中的consumer。broker的作用类似于缓存，即活跃的数据和离线处理系统之间的缓存。

部件

Partition

Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。

Topic

特指Kafka处理的消息源（feeds of messages）的不同分类。

Message

消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。

Producers

消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。

Consumers

消息和数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。

Broker

缓存代理，Kafa集群中的一台或多台服务器统称为broker。

消息发送图

流程图

Kafka如何搭建及创建topic、发送消息、消费消息？

Topics 和Logs

topic细节图

一个topic是对一组消息的归纳。对每个topic，Kafka 对它的日志进行了分区

log implementation

每个分区都由一系列有序的、不可变的消息组成，这些消息被连续的追加到分区中。分区中的每个消息都有一个连续的序列号叫做offset,用来在分区中唯一的标识这个消息。

实际上每个consumer唯一需要维护的数据是消息在日志中的位置，也就是offset.这个offset有consumer来维护：一般情况下随着consumer不断的读取消息，这offset的值不断增加，但其实consumer可以以任意的顺序读取消息，比如它可以将offset设置成为一个旧的值来重读之前的消息。

Producers

Producer将消息发布到它指定的topic中,并负责决定发布到哪个分区。

通常简单的由负载均衡机制随机选择分区，但也可以通过特定的分区函数选择分区。使用的更多的是第二种。

Consumers

队列模式和发布-订阅模式

队列模式中，consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到。

发布-订阅模式中消息被广播到所有的consumer中。

Consumers可以加入一个consumer 组，共同竞争一个topic，topic中的消息将被分发到组中的一个成员中。

如果所有的consumer都在一个组中，这就成为了传统的队列模式，在各consumer中实现负载均衡。

如果所有的consumer都不在不同的组中，这就成为了发布-订阅模式，所有的消息都被分发到所有的consumer中。

集群关系图

优点

相比传统的消息系统，Kafka可以很好的保证有序性。

Kafka可以在多个consumer组并发的情况下提供较好的负载均衡。

环境搭建

下载解压

https://www.apache.org/dyn/closer.cgi?path=/kafka//kafka_.tgz

tar -xzf kafka_2.9.2-0.8.1.1.tgz

cd kafka_2.9.2-0.8.1.1

启动服务

Kafka用到了Zookeeper，所有首先启动Zookper。

bin/zookeeper-server-start.sh config/zookeeper.properties

启动kafka

bin/kafka-server-start.sh config/server.properties

测试

创建topic

创建一个叫做“test”的topic，它只有一个分区，一个副本。

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

通过list命令查看创建的topic

bin/kafka-topics.sh --list --zookeeper localhost:2181

发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动consumer

Kafka也有一个命令行consumer可以读取消息并输出到标准输出：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

kafka开发环境搭建

maven项目

org.apache.kafka kafka_2.10 0.8.0

参数配置

package com.sohu.kafkademon; public interface KafkaProperties { final static String zkConnect = '10.22.10.139:2181'; final static String groupId = 'group1'; final static String topic = 'topic1'; final static String kafkaServerURL = '10.22.10.139'; final static int kafkaServerPort = 9092; final static int kafkaProducerBufferSize = 64 * 1024; final static int connectionTimeOut = 20000; final static int reconnectInterval = 10000; final static String topic2 = 'topic2'; final static String topic3 = 'topic3'; final static String clientId = 'SimpleConsumerDemoClient'; }

producer，consumer等

学习总结来源

http://www.aboutyun.com/thread-12882-1-1.html

http://blog.jobbole.com/75328/

http://www.cnblogs.com/likehua/p/3999538.html

使用模型

如何决定发送到哪一个分区

生产者可以指定key 和分区的负载策略

key

实现kafka提供的分区路由接口

客户端回自动执行该接口, 在生产者发送到 kafka server 时指定分区

如何实现广播

topic反正只有一个,消息也只有一份

kafka保证多个consumer group 中只有一个consumer消费到该消息多个consumer group 可多次消费该topic消息

不同的consumer 消费不同的分区

Kafka 对消息的分配是以 Partition 为单位分配的

无法保证同一个 Consumer Group 里的 Consumer 均匀消费数据

同一个 Partition 里的数据是有序的，这种设计可以保证每个 Partition 里的数据可以被有序消费。

consumer 上下线会被动态分配

consumer rebalance?

high level consumer 的设计

其他mq 通过记录消息的消费状态, 并且通过将多个消息路由到多个队列实现的广播.

消费端offset处理

指定offset消费

重置某个分区的offset 针对某一个consumer

consumer会被kafka感知到

seekToEnd

seekToBeginning

offset可以被放在第三方,例如数据库, 这样可以指定消费者的消费位置......

每个分区的offset都该被记住, 这样即使reblance 或者重新订阅从指定offset开始订阅

通过wakeup 可以让poll方法响应中断

consumer可以捕获reblance事件, 再丢失之后,可以重新提交最新的offset 被分配也能得到最新的offset

一个consumer group 和一个分区有一个offset

每一个group 都会存在一个协调者,负责consumer过期等问题(重平衡)

重平衡期间不会消费消息

代码示例

http://www.dengshenyu.com/%E5%88%86%E5%B8%83%E5%BC%8F%E7%B3%BB%E7%BB%9F/2017/11/14/kafka-consumer.html

consumer

特性

consumer group 由 broker server 管理,不由zk管理

可完全不依赖于zk

consumer 直接连接到group 的Coordinator

1. 连接到最后一次连接的broker（如果是刚启动的consumer，则要根据配置中的borker）。它会响应一个包含coordinator信息(host, port等)的response。 2）连接到coordinator。

创建时指定group

consumer client 非线程安全

client自己不实现线程模型,而是由客户端自己poll,自己实现线程模型

分支主题

reblance

过程

1）会给一个coordinator发起Join请求（请求中要包括自己的一些元数据，例如自己感兴趣的topics）

2）Coordinator 根据这些consumer的join请求，选择出一个leader，并通知给各个consumer。这里的leader是consumer group 内的leader，是由某个consumer担任，不要与partition的leader混淆。

3）Consumer leader 根据这些consumer的metadata，重新为每个consumer member重新分配partition。分配完毕通过coordinator把最新分配情况同步给每个consumer。

4）Consumer拿到最新的分配后，继续工作。

offset 如何管理?

kafka server不管offset

需要consumer 自己管理offset

存储地方

参与者

Topic

每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 Topic

物理上不同 Topic 的消息分开存储

Partition

每个 Topic 包含一个或多个 Partition.

为了提高吞吐, 每个topic都被分成多个分区.

单独一个目录(可分散IO)

实际的消息持久化格式

message length ： 4 bytes (value: 1+4+n) 'magic' value ： 1 byte crc ： 4 bytes payload ： n bytes

可在配置文件指定topic默认分区数, 也可以在创建topic时指定

Producer负责发布消息到 Kafka broker

Consumer消息消费者，向 Kafka broker 读取消息的客户端

Consumer Group每个 Consumer 属于一个特定的 Consumer Group

可指定重复消费消息, 采用pull 模型,从offset拉取数据

能得到消息的分区

zookeeper

kafka server 选主

为什么需要主??

记录consumer消费的 offset

是否是性能瓶颈

pull 还是push

push

消费者处于被动. 容易处理不过来

broker 需要记录消费者的消费状况,例如多少个消息没有被ack.

推送更加及时

pull

Consumer 的消费能力以适当的速率消费消息

broker 实现更加轻量, 消费者自己记录自己的状态

如何保证消费者重启时, 不重复消费

持久化实现

分区的实现

每一个分区都在一个目录上

可以分散IO

不同的目录可以挂载不同的磁盘

一个分区也会存在多个.kafka后缀的文件

有单独的索引文件,指定不同.kafka文件的offset范围

分区文件过期策略

kafka不删除已经被消费的数据

一是基于时间，二是基于 Partition 文件大小

只会清理被消费的数据

kafka是否不记录消息的消费记录

CAP支持

如何对消息冗余

避免一台机器挂了, 消息不能被消费

参考 zk的特性, 只会有一个master负责读写, 其他只是同步数据.

为啥其他备份不负责读? 难道是延迟吗?

同步的单位是分区

所以一个分区需要一个master

备份算法

将所有 Broker（假设共 n 个 Broker）和待分配的 Partition 排序

将第 i 个 Partition 分配到第（i mod n）个 Broker 上

将第 i 个 Partition 的第 j 个 Replica 分配到第（(i + j) mode n）个 Broker 上

ISR（in-sync replicas

其保持同步的 Replica 列表

follower 的消息落后于leader太多

已被废弃

消息量太大时,容易出bug

follower 多久没有fetch消息

High Wartermark

leader 回将自己受到的msg的offset 置为最新的HW

Commit Offset 标记的是最新的可被消费的

为什么使用isr

由于 Leader 可移除不能及时与之同步的 Follower，故与同步复制相比可避免最慢的 Follower 拖慢整体速度，也即 ISR 提高了系统可用性

ISR 中的所有 Follower 都包含了所有 Commit 过的消息，而只有 Commit 过的消息才会被 Consumer 消费，故从 Consumer 的角度而言，ISR 中的所有 Replica 都始终处于同步状态，从而与异步复制方案相比提高了数据一致性。

ISR 可动态调整，极限情况下，可以只包含 Leader，极大提高了可容忍的宕机的 Follower 的数量。与 Majority Quorum 方案相比，容忍相同个数的节点失败，所要求的总节点数少了近一半。

min.insync.replicas 参数指定了 Broker 所要求的 ISR 最小长度，默认值为 1

对于消费者 isr 中的follower 都具有commit的消息, 但是对于producer ,端的成功并不一定是说 isr 所有都ack 才会成功, 而是可以指定的, 必要时,生产者可不关系这个阶段的消息可用性

Producer 可以通过 acks 参数指定最少需要多少个 Replica 确认收到该消息才视为该消息发送成功。acks 的默认值是 1，即 Leader 收到该消息后立即告诉 Producer 收到该消息

此时如果在 ISR 中的消息复制完该消息前 Leader 宕机，那该条消息会丢失。而如果将该值设置为 0，则 Producer 发送完数据后，立即认为该数据发送成功，不作任何等待，而实际上该数据可能发送失败，并且 Producer 的 Retry 机制将不生效

根据场景决定生产端的可用性, 大数据场景吞吐量优先. 可以考虑设置为0

如何选举

Majority Vote

少数服务多数 zk采用的

2f+1 的节点备份数据必须成功f+1 个

最多容忍f个节点挂了

只要有一个follower在isr里就行

该场景下可能会有潜在的数据丢失(ISR 列表并不一定是最新的)

如果isr为空,都挂了咋办

等待 ISR 中的任一个 Replica“活”过来，并且选它作为 Leader

选择第一个“活”过来的 Replica（不一定是 ISR 中的）作为 Leader

默认

kafka会有一个master , 负责分区的选举和分区备份分配

通过rpc 通知

避免分区数量太多, zk 管理不过来

IO优化

pagecache

使用pagecache 优点

I/O Scheduler 会将连续的小块写组装成大块的物理写从而提高性能

- I/O Scheduler 会尝试将一些写操作重新按顺序排好，从而减少磁盘头的移动时间

- 充分利用所有空闲内存（非 JVM 内存）。如果使用应用层 Cache（即 JVM 堆内存），会增加 GC 负担

- 读操作可直接在 Page Cache 内进行。如果消费和生产速度相当，甚至不需要通过物理磁盘（直接通过 Page Cache）交换数据

- 如果进程重启，JVM 内的 Cache 会失效，但 Page Cache 仍然可用

每个分区会对应多个segement

缺点

机器宕机时，Page Cache 内的数据未写入磁盘从而造成数据丢失。但是这种丢失只发生在机器断电等造成操作系统不工作的场景

可以由 Kafka 层面的 Replication 机制去解决。如果为了保证这种情况下数据不丢失而强制将 Page Cache 中的数据 Flush 到磁盘，反而会降低性能。也正因如此，Kafka 虽然提供了 flush.messages 和 flush.ms 两个参数将 Page Cache 中的数据强制 Flush 到磁盘，但是 Kafka 并不建议使用

内存零拷贝技术

message 从producer 开始压缩

可用性保障

生产者

如果同步发送. producer会在发送失败的时候, 重试发送

异步模式发送失败重试n次, 直接抛弃

在备份模式下

参考zk master 向follower 传播数据时,当有半数节点ack 则返回消息成功

follower不是等落磁盘才 ack, 而是收到立即ack(内存中)

赌不会存在两台机器同时crash

不能完全保证异常发生后该条消息一定能被 Consumer 消费

一个权衡.机会非常小