脚本宝典收集整理的这篇文章主要介绍了Kafka,脚本宝典觉得挺不错的,现在分享给大家,也给大家做个参考。
Apache Kafka 是一个开源分布式事件流平台,被数千家公司用于高性能数据管道、流分析、数据集成和任务关键型应用程序。
Kafka本质上是一个 MQ(Message Queue),使用消息队列的好处:
Kafka 存储的消息来自任意多被称为 Producer 生产者的进程。数据从而可以被发布到不同的 Topic 主题下的不同 Partition 分区。
在一个分区内,这些消息被索引并连同时间戳存储在一起。其它被称为 Consumer 消费者的进程可以从分区订阅消息。
Kafka 运行在一个由一台或多台服务器组成的集群上,并且分区可以跨集群结点分布。
Kafka 一些重要概念:
Kafka 中消息是以 Topic 进行分类的,生产者生产消息,消费者消费消息,面向的都是同一个 Topic。
Topic 是逻辑上的概念,而 Partition 是物理上的概念,每个 Partition 对应于一个 log 文件,该 log 文件中存储的就是 Producer 生产的数据。
Producer 生产的数据会不断追加到该 log 文件末端,且每条数据都有自己的 Offset。
消费者组中的每个消费者,都会实时记录自己消费到了哪个 Offset,以便出错恢复时,从上次的位置继续消费。
副本机制(Replication),也可以称之为备份机制,通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝。副本机制的好处在于:
目前Kafka只实现了副本机制带来的第 1 个好处,即是提供数据冗余实现高可用性和高持久性。
在kafka生产环境中,每台 Broker 都可能保存有各个主题下不同分区的不同副本,因此,单个 Broker上存有成百上千个副本的现象是非常正常的。
下图展示了一个有 3 台 Broker 的 Kafka 集群上的副本分布情况。从图中可以看到,主题 1 分区 0 的 3个副本分散在 3 台 Broker 上,其他主题分区的副本也都散落在不同的 Broker 上,从而实现数据冗余。
基于领导者(Leader-based)的副本机制
参数名称 | 描述 |
---|---|
bootstrap.servers | 生 产 者 连 接 集 群 所 需 的 broker 地 址 清 单 。 例 如 hadoop102:9092,hadoop103:9092,hadoop104:9092,可以 设置 1 个或者多个,中间用逗号隔开。注意这里并非需要所有的 broker 地址,因为生产者从给定的 broker 里查找到其他 broker 信息。 |
key.serializer 和 value.serializer | 指定发送消息的 key 和 value 的序列化类型。一定要写 全类名。 |
buffer.memory | RecordAccumulator 缓冲区总大小, 默认 32m。 |
batch.size | 缓冲区一批数据最大值, 默认 16k。适当增加该值,可 以提高吞吐量,但是如果该值设置太大,会导致数据 传输延迟增加。 |
linger.ms | 如果数据迟迟未达到 batch.size, sender 等待 linger.time 之后就会发送数据。单位 ms, 默认值是 0ms,表示没 有延迟。 生产环境建议该值大小为 5-100ms 之间。 |
acks | 0:生产者发送过来的数据,不需要等数据落盘应答。 1:生产者发送过来的数据, Leader 收到数据后应答。 -1(all):生产者发送过来的数据, Leader+和 isr 队列 里面的所有节点收齐数据后应答。 默认值是-1, -1 和 all 是等价的。 |
max.in.flight.requests.per.connection | 允许最多没有返回 ack 的次数, 默认为 5,开启幂等性 要保证该值是 1-5 的数字。 |
retries | 当消息发送出现错误的时候,系统会重发消息。 retries 表示重试次数。 默认是 int 最大值, 2147483647。 如果设置了重试,还想保证消息的有序性,需要设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候,其他的消息可能发送 成功了。 |
retry.backoff.ms | 两次重试之间的时间间隔,默认是 100ms。 |
enable.idempotence | 是否开启幂等性, 默认 true,开启幂等性。 |
compression.type | 生产者发送的所有数据的压缩方式。 默认是 none,也 就是不压缩。 支持压缩类型: none、 gzip、 snappy、 lz4 和 zstd。 |
producer就是生产者,是数据的入口。Producer在写入数据的时候永远的找leader,不会直接将数据写入follower。
Kafka 的消息组织方式实际上是三级结构:主题 - 分区 - 消息。主题下的每条消息只会保存在某一个分区中,而不会在多个分区中被保存多份。如下所示:
分区的作用主要提供负载均衡的能力,能够实现系统的高伸缩性(Scalability)。不同的分区能够被放置到不同节点的机器上,而数据的读写操作也都是针对分区这个粒度而进行的,这样每个节点的机器都能独立地执行各自分区的读写请求处理。这样,当性能不足的时候可以通过添加新的节点机器来增加整体系统的吞吐量。
所谓分区策略是决定生产者将消息发送到哪个分区的算法。Kafka 为我们提供了默认的分区策略,同时它也支持你自定义分区策略。
Round-robin 策略,即顺序分配。比如一个主题下有 3 个分区,那么第一条消息被发送到分区 0,第二条被发送到分区 1,第三条被发送到分区 2,以此类推。当生产第 4 条消息时又会重新开始,即将其分配到分区 0
轮询策略有非常优秀的负载均衡表现,它总是能保证消息最大限度地被平均分配到所有分区上,故默认情况下它是最合理的分区策略,也是我们最常用的分区策略之一。
随机就是我们随意地将消息放置到任意一个分区上,如下面这张图所示
Kafka 允许为每条消息定义消息键,简称为 Key。这个 Key 的作用非常大,它可以是一个有着明确业务含义的字符串,比如客户代码、部门编号或是业务 ID 等;也可以用来表征消息元数据。特别是在 Kafka不支持时间戳的年代,在一些场景中,工程师们都是直接将消息创建时间封装进 Key 里面的。一旦消息被定义了 Key,那么你就可以保证同一个 Key 的所有消息都进入到相同的分区里面,由于每个分区下的消息处理都是有顺序的,故这个策略被称为按消息键保序策略。
参数名称 | 描述 |
---|---|
bootstrap.servers | 向 Kafka 集群建立初始连接用到的 host/port 列表。 |
key.deserializer 和 value.deserializer | 指定接收消息的 key 和 value 的反序列化类型。一定要写全 类名。 |
group.id | 标记消费者所属的消费者组。 |
enable.auto.commit | 默认值为 true,消费者会自动周期性地向服务器提交偏移 量。 |
auto.commit.interval.ms | 如果设置了 enable.auto.commit 的值为 true, 则该值定义了 消费者偏移量向 Kafka 提交的频率, 默认 5s。 |
auto.offset.reset | 当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 (如,数据被删除了),该如何处理? earliest:自动重置偏 移量到最早的偏移量。 latest: 默认, 自动重置偏移量为最 新的偏移量。 none:如果消费组原来的( previous)偏移量 不存在,则向消费者抛异常。 anything:向消费者抛异常。 |
offsets.topic.num.partitions | __consumer_offsets 的分区数, 默认是 50 个分区。 |
heartbeat.interval.ms | Kafka 消费者和 coordinator 之间的心跳时间, 默认 3s。 该条目的值必须小于 session.timeout.ms ,也不应该高于 session.timeout.ms 的 1/3。 |
session.timeout.ms | Kafka 消费者和 coordinator 之间连接超时时间, 默认 45s。 超过该值,该消费者被移除,消费者组执行再平衡。 |
max.poll.interval.ms | 消费者处理消息的最大时长, 默认是 5 分钟。超过该值,该 消费者被移除,消费者组执行再平衡。 |
fetch.min.bytes | 默认 1 个字节。消费者获取服务器端一批消息最小的字节 数。 |
fetch.max.wait.ms | 默认 500ms。如果没有从服务器端获取到一批数据的最小字 节数。该时间到,仍然会返回数据。 |
fetch.max.bytes | 默认 Default: 52428800( 50 m)。消费者获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值 (50m)仍然可以拉取回来这批数据,因此,这不是一个绝 对最大值。一批次的大小受 message.max.bytes ( broker config) or max.message.bytes (topic config) 影响。 |
max.poll.records | 一次 poll 拉取数据返回消息的最大条数, 默认是 500 条。 |
Consumer 采用 Pull(拉取)模式从 Broker 中读取数据。
Pull 模式则可以根据 Consumer 的消费能力以适当的速率消费消息。Pull 模式不足之处是,如果 Kafka没有数据,消费者可能会陷入循环中,一直返回空数据。
因为消费者从 Broker 主动拉取数据,需要维护一个长轮询,针对这一点, Kafka 的消费者在消费数据时会传入一个时长参数 timeout。
如果当前没有数据可供消费,Consumer 会等待一段时间之后再返回,这段时长即为 timeout。
一个 Consumer Group 中有多个 Consumer,一个 Topic 有多个 Partition,所以必然会涉及到Partition 的分配问题,即确定哪个 Partition 由哪个 Consumer 来消费。
Kafka 有三种分配策略:
当消费者组内消费者发生变化时,会触发分区分配策略(方法重新分配)。这里主要讲Range、RoundRobin。
Range 方式是按照主题来分的,不会产生轮询方式的消费混乱问题。
假设我们有10个分区,3个消费者,排完序的分区将会是0,1,2,3,4,5,6,7,8,9;消费者线程排完序将会是C1-0,C2-0,C3-0。然后将partitions的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽,那么前面几个消费者线程将会多消费一个分区。
在我们的例子里面,
我们有10个分 区,3个消费者线程, 10/3 = 3,而且除不尽,那么消费者线程 C1-0将会多消费一个分区的结果看起来是这样的:
C1-0 将消费 0, 1, 2, 3 分区
C2-0将消费 4,5,6分区
C3-0将消费 7,8,9分区
假如我们有11个分区,那么最后分区分配的结果看起来是这样的:
C1-0将消费 0,1,2,3分区
C2-0将消费 4,5,6,7分区
C3-0 将消费 8, 9, 10 分区
假如我们有2个主题(T1和T2),分别有10个分区,那么最后分区分配的结果看起来是这样的:
C1-0 将消费 T1主题的 0, 1, 2, 3 分区以及 T2主题的 0, 1, 2, 3分区
C2-0将消费 T1主题的 4,5,6分区以及 T2主题的 4,5,6分区
C3-0将消费 T1主题的 7,8,9分区以及 T2主题的 7,8,9分区
可以看出,C1-0 消费者线程比其他消费者线程多消费了2个分区,这就是Range strategy的一个很明显的弊端
即是,如下图所示,Consumer0、Consumer1 同时订阅了主题 A 和 B,可能造成消息分配不对等问题,当消费者组内订阅的主题越多,分区分配可能越不均衡。
RoundRobin 轮询方式将分区所有作为一个整体进行 Hash 排序,消费者组内分配分区个数最大差别为1,是按照组来分的,可以解决多个消费者消费数据不均衡的问题。
轮询分区策略是把所有partition和所有consumer线程都列出来,然后按照hashcode进行排序。最后通过轮询算法分配partition给消费线程。如果所有consumer实例的订阅是相同的,那么partition会均匀分布。
在我们的例子里面,假如按照 hashCode排序完的topic-partitions组依次为T1-5,T1-3,T1-0,T1-8,T1-2,T1-1,T1-4,T1-7,T1-6,T1-9,我们的消费者线程排序为C1-0,C1-1,C2-0,C2-1,最后分区分配的结果为:
C1-0将消费 T1-5,T1-2,T1-6分区;
C1-1将消费 T1-3,T1-1,T1-9分区;
C2-0将消费 T1-0,T1-4分区;
C2-1将消费 T1-8,T1-7分区
但是,当消费者组内订阅不同主题时,可能造成消费混乱,如下图所示,Consumer0 订阅主题 A,Consumer1 订阅主题 B。
将 A、B 主题的分区排序后分配给消费者组,TopicB 分区中的数据可能分配到 Consumer0 中。
使用轮询分区策略必须满足两个条件:
为保证 Producer 发送的数据,能可靠地发送到指定的 Topic,Topic 的每个 Partition 收到 Producer发送的数据后,都需要向 Producer 发送 ACK(ACKnowledge 确认收到)。
如果 Producer 收到 ACK,就会进行下一轮的发送,否则重新发送数据。
何时发送 ACK?确保有 Follower 与 Leader 同步完成,Leader 再发送 ACK,这样才能保证 Leader 挂掉之后,能在 Follower 中选举出新的 Leader 而不丢数据。
多少个 Follower 同步完成后发送 ACK?全部 Follower 同步完成,再发送 ACK。
方案 | 优点 | 缺点 |
---|---|---|
半数以上完成同 步,就发送ack | 延迟低 | 选举新的 障,需要leader 2n+1个副本。 时,容忍n台节点故 |
全部完成同步, 才发送ack | 选举新的leader时,容忍n台节点 故障,需要n+1个副本。 | 延迟高 |
当采用第二种方案时,所有 Follower 完成同步,Producer 才能继续发送数据,设想有一个 Follower 因为某种原因出现故障,那 Leader 就要一直等到它完成同步。 这个问题怎么解决?
对于某些不太重要的数据,对数据的可靠性要求不是很高,能够容忍数据的少量丢失,所以没必要等ISR 中的 Follower 全部接受成功。
所以 Kafka 为用户提供了三种可靠性级别,用户根据可靠性和延迟的要求进行权衡,选择以下的配置。
ACK 参数配置:
没有一个中间件能够做到百分之百的完全可靠,可靠性更多的还是基于几个9的衡量指标,比如4个9、5个9。软件系统的可靠性只能够无限去接近100%,但不可能达到100%。所以kafka如何是实现最大可能的可靠性呢?
bin/kafka-topics.sh
参数 | 描述 |
---|---|
--bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
--topic <String: topic> | 操作的 topic 名称。 |
--create | 创建主题。 |
--delete | 删除主题。 |
--alter | 修改主题。 |
--list | 查看所有主题。 |
--describe | 查看主题详细描述。 |
--partitions <Integer: # of partitions> | 设置分区数。 |
--replication-factor<Integer: replication factor> | 设置分区副本。 |
--config <String: name=value> | 更新系统默认的配置。 |
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 3 --topic test
选项说明:
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test
bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic test --partitions 3
bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test
bin/kafka-console-producer.sh
参数 | 描述 |
---|---|
--bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
--topic <String: topic> | 操作的 topic 名称。 |
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
bin/kafka-console-consumer.sh
参数 | 描述 |
---|---|
--bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
--topic <String: topic> | 操作的 topic 名称。 |
--from-beginning | 从头开始消费。 |
--group <String: consumer group id> | 指定消费者组ID。 |
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic test
以上是脚本宝典为你收集整理的Kafka全部内容,希望文章能够帮你解决Kafka所遇到的问题。
本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ:384754419,请注明来意。