KAFKA基础知识

作者: 夏天嘚花花 | 来源:发表于2019-02-25 09:23 被阅读0次

kafka相关知识总结
Kafka & Zookeeper
Kafka为啥这么高吞吐
kafka 基础知识
kafka基础知识
KAFKA基础知识
kafka基础知识
SpringCloud-Stream-Kafka
kafka 学习笔记 3 - Java 使用 kafka 收发消
Kafka基础知识扫盲

分区与主题

kafka的消息是通过主题进行分类的。主题可以被分成若干个分区。其实主题是一个逻辑概念，主题在物理上被分成多个分区分别存储到不同的服务器上，如图。

image.png

下面创建一个主题看主题在KAFKA上是怎么存储的

//创建一个主题名叫topic20190224,包含partitions=5个分区,replication-factor=1一个副本。
sh kafka-topics.sh --create --zookeeper 192.168.199.128:2182 --replication-factor 1 --partitions 5 --topic topic20190224

使用命令查看topic20190224主题的分区情况

[root@server-1 bin]# sh kafka-topics.sh --describe --zookeeper 192.168.199.128:2182 --topic topic20190224
Topic:topic20190224 PartitionCount:5    ReplicationFactor:1 Configs:
    Topic: topic20190224    Partition: 0    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic20190224    Partition: 1    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic20190224    Partition: 2    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic20190224    Partition: 3    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic20190224    Partition: 4    Leader: 1   Replicas: 1 Isr: 1

通过上图我们知道topic_20190224有5个分区，我们取第二行进行分析:
Topic: topic20190224 Partition: 0 Leader: 3 Replicas: 3 Isr: 3
Partition: 0 : 表示这个是分区0的记录
Leader: 3 : 表示分区0的leader是在第三台服务器(由于我们这里只有一个副本，所以就没有follower)
Replicas: 3 : 表示分区0分布在哪几台机器上，这里如果有多个副本会用逗号隔开(2,3表示分区0分布在id=2,3这两台机器上)

生产者

生产者创建消息。在kafka中一个消息会被发布到特定的主题上，生产者在默认情况下会将消息均匀的分布到所有分区上，而不关心消息具体被写到哪个分区上。但在某些情况下可以通过消息建键和分区器来实现将消息发布到指定的分区上。

消费者

消费者订阅一个主题后会按照消息生成的顺序读取，并通过检查偏移量来区分哪些是读过的消息(消费者把每个分区最后读取的消息偏移量保存到Zookeeper上)。

消费者与消费群组

消费者是消费群组的一部分。群组保证每个分区只能被一个消费者使用。如图，3个消费者同属一个群组，主题一共有4个分区。这种情况下，这个消费者群组为了消费这个主题所有分区，必定有一个消费者得消费两个分区，且不能出现同一个群组的多个消费者同时消费同一个分区，也就是说一个分区只能被一个群组里的消费者消费

image.png

kafka消息保留(持久化)

默认设置
1.按照时间保留默认7天
2.按照字节数保留
当达到上限时消息会自动删除

kafka元数据

image.png
topics注册信息
例如获取myTopicTwo主题的信息

[zk: 192.168.199.128:2182(CONNECTED) 3] get /brokers/topics/myTopicTwo
{"version":1,"partitions":{"2":[3],"1":[2],"0":[1]}}
//以下是详解，可以看出这个主题有3个分区,2号分区的副本在3号机器上
{
    "version": "版本编号目前固定为数字1",
    "partitions": {
        "partitionId编号": [
            同步副本组brokerId列表
        ],
        "partitionId编号": [
            同步副本组brokerId列表
        ],
        .......
    }
}

partition状态信息

[zk: 192.168.199.128:2182(CONNECTED) 0] ls /brokers/topics/myTopicTwo/partitions
[0, 1, 2]
[zk: 192.168.199.128:2182(CONNECTED) 1] get /brokers/topics/myTopicTwo/partitions/0/state
{"controller_epoch":3,"leader":1,"version":1,"leader_epoch":0,"isr":[1]}
cZxid = 0x50000006c
ctime = Tue Sep 11 14:47:36 PDT 2018
//以下是详解
{
"controller_epoch": 表示kafka集群中的中央控制器选举次数,
"leader": 表示该partition选举leader的brokerId,
"version": 版本编号默认为1,
"leader_epoch": 该partition leader选举次数,
"isr": [同步副本组brokerId列表]
}

Broker注册信息
每个broker的配置文件中都需要指定一个数字类型的id(全局不可重复),此节点为临时znode(EPHEMERAL)

[zk: 192.168.199.128:2182(CONNECTED) 5] ls /brokers/ids   
[1, 2, 3]
[zk: 192.168.199.128:2182(CONNECTED) 6] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://192.168.199.128:9092"],"jmx_port":-1,"host":"192.168.199.128","timestamp":"1551277782382","port":9092,"version":4}
{
"jmx_port": jmx端口号,
"timestamp": kafka broker初始启动时的时间戳,
"host": 主机名或ip地址,
"version": 版本编号默认为1,
"port": kafka broker的服务端端口号,由server.properties中参数port确定
}

Controller epoch
/controller_epoch --> int (epoch)
此值为一个数字,kafka集群中第一个broker第一次启动时为1，以后只要集群中center controller中央控制器所在broker变更或挂掉，就会重新选举新的center controller，每次center controller变更controller_epoch值就会 + 1

[zk: 192.168.199.128:2182(CONNECTED) 7] get /controller_epoch
18
cZxid = 0x100000022

Controller注册信息

[zk: 192.168.199.128:2182(CONNECTED) 8] get /controller
{"version":1,"brokerid":1,"timestamp":"1551277782724"}
{
"version": 版本编号默认为1,
"brokerid": kafka集群中broker唯一编号,
"timestamp": kafka broker中央控制器变更时的时间戳
}

主题的配置信息
获取某一个主题的配置信息

[zk: 192.168.199.128:2182(CONNECTED) 10] get /config/topics/myTopicTwo
{"version":1,"config":{}}

副本复制

副本有两种类型:首领副本和跟随者副本
首领副本 : 每个分区只有一个,为保证一致性，生产者请求和消费者请求都经过这个副本
跟随者副本:不处理客户端的请求，唯一的作用是从首领副本复制消息保证一致性,若首领宕机,其中的一个会成为首领

kafka的存储结构

kafka的消息是以主题为基本单位的，主题又有多个分区构成。消息存储的路径通过log.dirs配置。在结构上消息对应一个Log对象，Log对象有划分成多个段，在物理上分成一个日志文件和两个索引文件(两个索引文件分别为偏移量索引文件和时间戳索引文件)。如图

image.png

0000000.log文件最大大小有log.segment.bytes决定，默认1GB,通过kafka又提供了按时间切分日志通过log.roll.ms或log.roll.hours设置,即使当没达到log.segment.bytes设定的值时，同样也能创建新的日志段

生产者性能测试

向主题为topic20190224发送10000(--num-records)条消息，每次消息字节大小(--record-size)为1000.

[root@server-1 bin]# sh kafka-producer-perf-test.sh --num-records 10000 --record-size 1000 --topic topic20190224 --throughput 1000 --producer-props bootstrap.servers=192.168.199.128:9092 acks=all
4969 records sent, 993.8 records/sec (0.95 MB/sec), 42.1 ms avg latency, 521.0 max latency.
10000 records sent, 997.605746 records/sec (0.95 MB/sec), 23.27 ms avg latency, 521.00 ms max latency, 3 ms 50th, 155 ms 95th, 321 ms 99th, 336 ms 99.9th.

测试结果字段;

image.png