Kafka架构深入

1. 消息队列

1.1 传统消息队列的应用场景

MQ传统应用场景之异步处理

1.2 消息队列的两种模式

1) 点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除)

  消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。

消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费

2)  发布/订阅模式（一对多，消费者消费数据之后不会清除消息）

  消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。

1.2 Kafka基础架构

• Producer: 消息生产者，就是向kafka broker 发送消息的客户端
• Consumer: 消息消费者，向kafka broker 取消息的客户端
• Consumer Group : 消费者组，由多个consumer组成，消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个消费者消费，消费者组之间互补影响。

所有的消费者都属于某个消费者组。

• Broker: 一个kafka服务器就是一个broker，一个集群由多个broker组成，一个broker可以容纳多个topic
• Topic: 可以理解一个队列
• Partition: 为了实现扩展性，一个非常大的topic可以分布到多个broker上，一个topic可以分为多个partition，每个partition 是一个有序的队列。
• Replica:  副本，为了保证集群的某个节点发生故障时，该节点上的partition 数据不丢失，且kafka仍然能继续工作，kafka提供了副本机制，一个topic的每个分区都有若干个副本，一个leader 和若干个follower
• Leader: 每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是leader
• Follower： 每个分区多个副本中的“从”，实时从leader中同步数据，保持和leader数据的同步，leader发生故障时，某个follower会成为新的leader

2. Kafka工作流程

Kafka中消息是以topic进行分类的，生产者生产消息，消费者消费消息，都是面向topic的。

topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。

3. Kafka 文件存储机制

由于生产者生产的消息会不断追加到log文件末尾，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment对应两个文件——“.index”文件和“.log”文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。例如，test这个topic有三个分区，则其对应的文件夹为test-0,test-1,test-2

-rw-rw-r--. 1 hadoop hadoop 10485760 4月  24 23:55 00000000000000000000.index
-rw-rw-r--. 1 hadoop hadoop        0 4月  24 23:55 00000000000000000000.log
-rw-rw-r--. 1 hadoop hadoop 10485756 4月  24 23:55 00000000000000000000.timeindex

index和log文件以当前segment的第一条消息的offset命名。下图为index文件和log文件的结构示意图

“.index”文件存储大量的索引信息，“.log”文件存储大量的数据，索引文件中的元数据指向对应数据文件中message的物理偏移地址。

3.1 kafka数据的存储位置和文件内容

　　查看kafka安装目录的配置文件config/server.properties

log.dirs=/opt/module/kafka_2.11-2.4.1/logs

  创建一个有三个分区，2个副本的 test主题

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

[hadoop@hadoop1 logs]$ ls /opt/module/kafka_2.11-2.4.1/logs/
test-1
test-2
[hadoop@hadoop2 logs]$ ls /opt/module/kafka_2.11-2.4.1/logs/
test-0
test-2
[hadoop@hadoop3 ~]$ ls /opt/module/kafka_2.11-2.4.1/logs/
test-0
test-1

查看segment file

[hadoop@hadoop1 logs]$ cd test-1/
[hadoop@hadoop1 test-1]$ ll
总用量 8
-rw-rw-r--. 1 hadoop hadoop 10485760 4月  24 23:55 00000000000000000000.index
-rw-rw-r--. 1 hadoop hadoop      147 4月  25 00:10 00000000000000000000.log
-rw-rw-r--. 1 hadoop hadoop 10485756 4月  24 23:55 00000000000000000000.timeindex

查看log文件

[hadoop@hadoop1 test-1]$ kafka-run-class.sh  kafka.tools.DumpLogSegments --files  00000000000000000000.log  --print-data-log

Dumping 00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 0 CreateTime: 1682352602628 size: 79 magic: 2 compresscodec: NONE crc: 1291193145 isvalid: true
| offset: 0 CreateTime: 1682352602628 keysize: -1 valuesize: 11 sequence: -1 headerKeys: [] payload: kafka flume
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 79 CreateTime: 1682352614290 size: 68 magic: 2 compresscodec: NONE crc: 2089698590 isvalid: true
| offset: 1 CreateTime: 1682352614290 keysize: -1 valuesize: 0 sequence: -1 headerKeys: [] payload:

payload:为消息体 

查看index文件

[hadoop@hadoop1 test-1]$ kafka-run-class.sh  kafka.tools.DumpLogSegments --files 00000000000000000000.index  --print-data-log
Dumping 00000000000000000000.index
offset: 0 position: 0

查看timeindex文件

[hadoop@hadoop1 test-1]$ kafka-run-class.sh  kafka.tools.DumpLogSegments --files 00000000000000000000.timeindex  --print-data-log
Dumping 00000000000000000000.timeindex
timestamp: 0 offset: 0
Found timestamp mismatch in :/opt/module/kafka_2.11-2.4.1/logs/test-1/00000000000000000000.timeindex
  Index timestamp: 0, log timestamp: 1682352602628