Kafka相关知识-编程知识

一、kafka架构

Kafka基础知识

Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多生产者、多订阅者，基于zookeeper协调的分布式日志系统(也可以当做MQ系统)，常见可以用于webynginx日志、访问日志，消息服务等等，Linkedin于 2010年贡献给了Apache基金会并成为顶级开源项目。主要应用场景是:日志收集系统和消息系统。

Kafka主要设计目标如下:

以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。

高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。支持KafkaServer间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。同时支持离线数据处理和实时数据处理。支持在线水平扩展

kafka是一种发布-订阅模式, 对于消息中间件，消息分推拉两种模式。Kafka只有消息的拉取，没有推送，可以通过轮询实现消息的推送。1.Kafka在一个或多个可以跨越多个数据中心的服务器上作为集群运行。

2.Kafka集群中按照主题分类管理，一个主题可以有多个分区，一个分区可以有多个副本分区。

3.每个记录由一个键，一个值和一个时间戳组成。

Kafka具有四个核心API:

1.ProducerAPI:允许应用程序将记录流发布到一个或多个Kafka主题。2.ConsumerAPI:允许应用程序订阅一个或多个主题并处理为其生成的记录流。3.StreamsAPI:允许应用程序充当流处理器，使用一个或多个主题的输入流，并生成一个或多个输出主题的输出流，从而有效地将输入流转换为输出流。4.ConnectorAPI:允许构建和运行将Kafka主题连接到现有应用程序或数据系统的可重用生产者或使用者。例如，关系数据库的连接器可能会捕获对表的所有更改。

Kafka优势

1.高吞吐量:单机每秒处理几十上百万的消息量。即使存储了许多TB的消息，它也保持稳定的性能。2.高性能:单节点支持上千个客户端，并保证零停机和零数据丢失。3.持久化数据存储:将消息持久化到磁盘。通过将数据持久化到硬盘以及replication防止数据丢失。

4.分布式系统，无需停机就可扩展机器。

5.可靠性-kafka是分布式，分区，复制和容错的。

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

7.支持online和offline的场景。

8.支持多种客户端语言。Kafka支持Java、.NET、PHP、Python等多种语言。

Kafka应用场景

日志收集：一个公司可以用Kafka可以收集各种服务的Log，通过Kafka以统一接口服务的方式开放给各种Consumer。

消息系统：解耦生产者和消费者、缓存消息等。

用户活动跟踪：用来记录web用户或者APP用户的各种活动，如网页搜索、搜索、点击，用户数据收集然后进行用户行为分析。

运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比

如报警和报告；

流式处理：比如Spark Streaming和Storm。

Kafka基本架构

消息和批次

Kafka的数据单元称为消息。可以把消息看成是数据库里的一个“数据行”或一条“记录”，消息由字节数组组成。批次就是一组消息，这些消息属于同一个主题和分区。

模式

消息模式（schema）有许多可用的选项，以便于理解。如JSON和XML，但是它们缺乏强类型处理能力。Kafka的

许多开发者喜欢使用Apache Avro。Avro提供了一种紧凑的序列化格式，模式和消息体分开。当模式发生变化时，不需要重新生成代码，它还支持强类型和模式进化，其版本既向前兼容，也向后兼容。

主题和分区

Kafka的消息通过主题进行分类。主题可比是数据库的表或者文件系统里的文件夹。主题可以被分为若干分区，一

个主题通过分区分布于Kafka集群中，提供了横向扩展的能力。

生产者和消费者

生产者创建消息。消费者消费消息。消息被发布到一个特定的主题上。、

Borker和集群

一个独立的Kafka服务器称为broker。broker接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保

存。broker为消费者提供服务，对读取分区的请求做出响应，返回已经提交到磁盘上的消息。单个broker可以轻松处理数千个分区以及每秒百万级的消息量。每个集群都有一个broker是集群控制器。

Kafka核心概念

Producer

生产者创建消息。

Consumer

消费者读取消息

Broker

一个独立的Kafka 服务器被称为broker，是集群的组成部分。

Topic

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

Partition

1. 主题可以被分为若干个分区，一个分区就是一个提交日志。

2. 消息以追加的方式写入分区，然后以先入先出的顺序读取。

3. 无法在整个主题范围内保证消息的顺序，但可以保证消息在单个分区内的顺序。

4. Kafka 通过分区来实现数据冗余和伸缩性。

5. 在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。

Replicas

kafka 使用主题来组织数据，每个主题被分为若干个分区，每个分区有多个副本。

Offset

生产者Offset：消息写入的时候，每一个分区都有一个offset，这个offset就是生产者的offset，同时也是这个分区的最新最大的offset。

消费者Offset：某个分区的offset情况，生产者写入的offset是最新最大的值是12，而当Consumer A进行消费时，从0开始消费，一直消费到了9，消费者的offset就记录在9，Consumer B就纪录在了11。

副本

Kafka通过副本保证高可用。副本分为首领副本(Leader)和跟随者副本(Follower)。

分区中的所有副本统称为AR（Assigned Repllicas），AR=ISR+OSR。

ISR

所有与leader副本保持一定程度同步的副本（包括Leader）组成ISR（In-Sync Replicas），ISR集合是AR集合中

的一个子集。

OSR

与leader副本同步滞后过多的副本（不包括leader）副本，组成OSR(Out-Sync Relipcas)。

Kafka的安装和配置

第一步：jdk安装，上传jdk-8u261-linux-x64.rpm到服务器并安装。

rpm -ivh jdk-8u261-linux-x64.rpm

第二步：配置java环境变量

vim /etc/profile

# 生效

source /etc/profile

# 验证

java -version

第三步：上传zookeeper安装包并解压。

tar -zxf zookeeper-3.4.14.tar.gz

cd /zookeeper-3.4.14/conf

# 复制zoo_sample.cfg命名为zoo.cfg

cp zoo_sample.cfg zoo.cfg

# 编辑zoo.cfg文件

vim zoo.cfg

dataDir=/usr/local/zookeeper/zookeeper-3.4.14/data

第四步：配置zookeeper环境变量

vim /etc/profile

启动命令：zkServer.sh start 查看状态命令：zkServer.sh status

第五步：上传kafka_2.12-1.0.2.tgz到服务器并解压

tar -zxf kafka_2.12-1.0.2.tgz

第六步：配置kafka环境变量

vim /etc/profile

第七步：修改kafka配置文件，连接zookeeper

# 进入配置文件夹修改server.properties文件

cd config/

vim server.properties

第七步：启动kafka

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

消费和主题

# 列出现有的主题

[root@node1 ~]# kafka-topics.sh --list --zookeeper localhost:2181/myKafka

# 创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。

[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_1 --partitions 1 --replication-factor 1

# 查看分区信息

[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --list

# 查看指定主题的详细信息

[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_1

# 删除指定主题

[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_1

# 开启生产者

[root@node1 ~]# kafka-console-producer.sh --topic topic_1 --broker-list localhost:9020

# 开启消费者

[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1

# 开启消费者方式二，从头消费，不按照偏移量消费

[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1 --from-beginning

Kafka消息接收和发送

生产者主要的对象有：KafkaProducer ，ProducerRecord 。

其中KafkaProducer 是用于发送消息的类，ProducerRecord 类用于封装Kafka的消息。

KafkaProducer 的创建需要指定的参数和含义：

二、Kafka高级特性

生产者

消息发送

流程图：

1. Producer创建时，会创建一个Sender线程并设置为守护线程。

2. 生产消息时，内部其实是异步流程；生产的消息先经过拦截器->序列化器->分区器，然后将消息缓存在缓冲

区（该缓冲区也是在Producer创建时创建）。

3. 批次发送的条件为：缓冲区数据大小达到batch.size或者linger.ms达到上限，哪个先达到就算哪个。

4. 批次发送后，发往指定分区，然后落盘到broker；如果生产者配置了retrires参数大于0并且失败原因允许重

试，那么客户端内部会对该消息进行重试。

5. 落盘到broker成功，返回生产元数据给生产者。

6. 元数据返回有两种方式：一种是通过阻塞直接返回，另一种是通过回调返回。

配置参数：

序列化器

Kafka中的数据都是字节数组，将消息发送到Kafka之前需要先将数据序列化为字节数组，序列化器的作用就是用于序列化要发送的消息。

分区器

默认分区计算：

1. 如果record提供了分区号，则使用record提供的分区号

2. 如果record没有提供分区号，则使用key的序列化后的值的hash值对分区数量取模

3. 如果record没有提供分区号，也没有提供key，则使用轮询的方式分配分区号。

拦截器

Producer拦截器（interceptor）和Consumer端Interceptor是在Kafka 0.10版本被引入的，主要用于实现Client

端的定制化控制逻辑。

Intercetpor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：

onSend(ProducerRecord)：该方法封装进KafkaProducer.send方法中，即运行在用户主线程中。Producer

确保在消息被序列化以计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修

改消息所属的topic和分区，否则会影响目标分区的计算。

onAcknowledgement(RecordMetadata, Exception)：该方法会在消息被应答之前或消息发送失败时调用，

并且通常都是在Producer回调逻辑触发之前。onAcknowledgement运行在Producer的IO线程中，因此不

要在该方法中放入很重的逻辑，否则会拖慢Producer的消息发送效率。

close：关闭Interceptor，主要用于执行一些资源清理工作。

原理

主线程：负责消息创建，拦截器，序列化器，分区器等操作，并将消息追加到消息收集器。

Sender线程：

该线程从消息收集器获取缓存的消息，将其处理为 <Node, List<ProducerBatch> 的形式，表示集群的broker节点。

进一步将<Node, List<ProducerBatch>转化为<Node, Request>形式，此时才可以向服务端发送数据。在发送之前，Sender线程将消息以 Map<NodeId, Deque<Request>> 的形式保存到InFlightRequests 中进行缓存，可以通过其获取 leastLoadedNode ,即当前Node中负载压力最小的一个，以实现消息的尽快发出。

消费者

消费组

消费者从订阅的主题消费消息，消费消息的偏移量保存在Kafka的名字是 __consumer_offsets 的主题中，消费者还可以将自己的偏移量存储到Zookeeper，需要设置offset.storage=zookeeper。推荐使用Kafka存储消费者的偏移量。因为Zookeeper不适合高并发。多个从同一个主题消费的消费者可以加入到一个消费组中。消费组中的消费者共享group_id。

消费者四种情况：

1. 消费组均衡地给消费者分配分区，每个分区只由消费组中一个消费者消费。

2. 消费组均衡地给消费者分配分区，每个分区只由消费组中一个消费者消费。

3. 如果在消费组中添加一个消费者2，则每个消费者分别从两个分区接收消息。

4. 如果消费组有四个消费者，则每个消费者可以分配到一个分区。

5. 如果向消费组中添加更多的消费者，超过主题分区数量，则有一部分消费者就会闲置，不会接收任何消息。

心跳机制

消费者宕机，退出消费组，触发再平衡，重新给消费组中的消费者分配分区。

Kafka 的心跳是 Kafka Consumer 和 Broker 之间的健康检查，只有当 Broker Coordinator 正常时，Consumer

才会发送心跳。broker 处理心跳的逻辑在 GroupCoordinator 类中：如果心跳超期， broker coordinator 会把消费者从 group中移除，并触发 rebalance。

Topic，Kafka用于分类管理消息的逻辑单元，类似与MySQL的数据库。

Partition，是Kafka下数据存储的基本单元，这个是物理上的概念。同一个topic的数据，会被分散的存储到多个partition中，这些partition可以在同一台机器上，也可以是在多台机器上。优势在于：有利于水平扩展，避免单台机器在磁盘空间和性能上的限制，同时可以通过复制来增加数据冗余性，提高容灾能力。为了做到均匀分布，通常partition的数量通常是Broker Server数量的整数倍。

Consumer Group，同样是逻辑上的概念，是Kafka实现单播和广播两种消息模型的手段。保证一个消费组获取到特定主题的全部的消息。在消费组内部，若干个消费者消费主题分区的消息，消费组可以保证一个主题的每个分区只被消费组中的一个消费者消费。

consumer 采用 pull 模式从 broker 中读取数据。

采用 pull 模式，consumer 可自主控制消费消息的速率，可以自己控制消费方式（批量消费/逐条消费)，还可以选择不同的提交方式从而实现不同的传输语义。

反序列化

Kafka的broker中所有的消息都是字节数组，消费者获取到消息之后，需要先对消息进行反序列化处理，然后才能

交给用户程序消费处理。消费者的反序列化器包括key的和value的反序列化器。

key.deserializer

value.deserializer

IntegerDeserializer

StringDeserializer

需要实现 org.apache.kafka.common.serialization.Deserializer<T> 接口。

位移提交

1. Consumer需要向Kafka记录自己的位移数据，这个汇报过程称为提交位移(Committing Offsets)

2. Consumer 需要为分配给它的每个分区提交各自的位移数据

3.位移提交的由Consumer端负责的，Kafka只负责保管。

4.位移提交分为自动提交和手动提交

5.位移提交分为同步提交和异步提交

消费位移管理

Kafka中，消费者根据消息的位移顺序消费消息。消费者的位移由消费者管理，可以存储于zookeeper中，也可以存储于Kafka主题__consumer_offsets中。Kafka提供了消费者API，让消费者可以管理自己的位移。

再均衡

重平衡其实就是一个协议，它规定了如何让消费者组下的所有消费者来分配topic中的每一个分区。比如一个topic

有100个分区，一个消费者组内有20个消费者，在协调者的控制下让组内每一个消费者分配到5个分区，这个分配的过程就是重平衡，是kafka为人诟病最多的一个点。

重平衡的触发条件主要有三个：

1. 消费者组内成员发生变更，这个变更包括了增加和减少消费者，比如消费者宕机退出消费组。

2. 主题的分区数发生变更，kafka目前只支持增加分区，当增加的时候就会触发重平衡。

3. 订阅的主题发生变化，当消费者组使用正则表达式订阅主题，而恰好又新建了对应的主题，就会触发重平衡。

重平衡过程中，消费者无法从kafka消费消息，这对kafka的TPS影响极大，而如果kafka集内节点较多，比如数百个，那重平衡可能会耗时极多。数分钟到数小时都有可能，而这段时间kafka基本处于不可用状态。所以在实际环境中，应该尽量避免重平衡发生。避免重平衡，是不可能，因为你无法完全保证消费者不会故障。而消费者故障是最常见引发重平衡的地方，需要尽力避免消费者故障，比如合理利用心跳来维持。控制发送心跳的频率，频率越高越不容易被误判。

消费者管理

消费组：consumer group是kafka提供的可扩展且具有容错性的消费者机制。

三个特性：

1. 消费组有一个或多个消费者，消费者可以是一个进程，也可以是一个线程

2. group.id是一个字符串，唯一标识一个消费组

3. 消费组订阅的主题每个分区只能分配给消费组一个消费者。

消费者位移：消费者在消费的过程中记录已消费的数据，即消费位移（offset）信息。

kafka提供了5个协议来处理与消费组协调相关的问题：

Heartbeat请求：consumer需要定期给组协调器发送心跳来表明自己还活着

LeaveGroup请求：主动告诉组协调器我要离开消费组

SyncGroup请求：消费组Leader把分配方案告诉组内所有成员

JoinGroup请求：成员请求加入组