kafka学习笔记

“我翻开招聘APP一查，这上面没有功能测试，歪歪斜斜的每页上都写着“大数据”几个字。我横竖睡不着，仔细看了半夜，才从字缝里看出字来，满本都写着两个字是“kafka”！”

那么让我们揭开kafka那神秘的面纱吧！了解什么事务之前我习惯于问自己几个问题：什么是kafka？为啥都要用kafka呢？它到底有什么魔力值得广大企事业和程序猿去选择它？

kafka是什么？

Apache Kafka是一个开放源代码的分布式事件流平台，成千上万的公司使用它来实现高性能数据管道，流分析，数据集成和关键任务等相关的应用程序。

嗯？说了但是又完全说…

那么它既然能被成千上万家公司信赖的理由是什么呢？

kafka的优势

总结来说就是：

• Kafka最大的特点就是收发消息非常快，Kafka每秒可以处理十万级别的消息，最低延迟只有几毫秒；

• 高可用性：允许集群中的节点失败，某个节点宕机，Kafka集群也能够正常工作

• 可靠性：Kafka能够允许数据的持久化存储，消息被持久化到磁盘，并支持数据备份防止数据丢失

• 高扩展性：一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量

那么kafka都被成千上万家公司用在什么场景上面呢？

kafka的使用场景

1. 构造实时流数据管道，它可以在系统或应用之间可靠地获取数据 (相当于message queue)，特别是在集群情况下，多个服务器需要建立交流

2. 构建实时流式应用程序，对这些流数据进行转换或者影响。(就是流处理，通过kafka stream topic和topic之间内部进行变化)

Producers：它允许应用程序将消息数据上发到Kafka集群中。

Consumers：它允许应用程序将消息数据从Kafka集群中拉取出来。

Connectors：Kafka的连接器可以连接到应用程序的生产者或者消费者的数据系统例如数据库。

Stream Processors：流处理器可以Kafka中拉取数据，也可以将数据写入到Kafka中。

上面说到了Producers、Consumers等等之流，这又是啥啊？话已至此不得不掏出祖传图了

kafka的系统架构

看着是不是有点眼花？没关系我们一点点来

• Producers：消息生产者，就是向主题发送消息的客户端

• Broker：一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topic

• Topic：可以理解为消息的分类，生产者和消费者面向的都是一个 topic，kafka的数据就保存在topic

• Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服 务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列，分区的作用是做负载，提高kafka的吞吐量

• Replication:每一个分区都有多个副本，一个 Leader 和若干个 Follower

• Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 Leader

• Follower：每个分区多个副本中的“从”，实时从 Leader 中同步数据，保持和 Leader 数据的同步。Leader 发生故障时，某个 Follower 会成为新的 Leader

• Consumer：消息消费者，订阅主题消息的客户端

• Consumer Group（CG）：消费者组由多个 consumer 组成。消费者组内每个消 费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者

扩展：

那么leader是怎么来的呢？

Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。Controller 的信息同步工作是依赖于 Zookeeper 。

Partition的结构：

理论太枯燥了，让我们动起小手玩一玩会好一点！

kafka部署

修改配置项

broker.id是broker 的全局唯一编号，不能重复，只能是数字listeners=PLAINTEXT://和advertised.listeners=PLAINTEXT://按需填写你自己主机和端口log.dirs按需填写你自己的主机日志路径

修改环境变量

/etc/profile新增环境变量

#KAFKA_HOME export KAFKA_HOME=/opt/module/kafka export PATH=$PATH:$KAFKA_HOME/bin#刷新环境变量source /etc/profile

启动kafka

# 启动ZooKeeper zkServer.sh start#启动kafkabin/kafka-server-start.sh -daemon config/server.properties

启动之后就可以愉快的玩耍啦

kafka命令行操作

生产者：

bin/kafka-console-producer.sh

--topic    操作的topic名称--bootstrap-server     连接kafka broker的主机名和端口号

topic相关：

bin/kafka-topics.sh

--topic    操作的topic名称--bootstrap-server    连接kafka broker的主机名和端口号--create    增加topic--delete    删除topic--alter    修改topic--list    查看topic列表--describe    查看topic详--partitions    设置分区数--replication-factor    设置分区副本

消费者：

bin/kafka-console-consumer.sh

--topic    操作的topic名称--bootstrap-server    连接kafka broker的主机名和端口号--from-beginning    查看主题数据详情

在我们可以具体操作，对kafka有更直观的感受后，还有很多问题等待我们去探索

有那么多分区，生产者怎么知道往哪个分区发呢？

生产者分区策略

（1）指明partition的情况下，直接将指明的值作为partition值；例如partition=0，所有数据写入分区0；

（2）没有指明partition值但有key的情况下，将key的hash值与topic的

partition数进行取余得到partition值；例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。

（3）既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）

知道往哪个分区发送数据后生产者如何提高吞吐量呢？

batch.size：批次大小，默认 16Klinger.ms：等待时间，修改为5-100mscompression.type：压缩snappyRecordAccumulator：缓冲区大小，修改为64m

在我们提高吞吐量的同时，如何保证数据不丢失即数据可靠性呢

数据可靠性

利用ack应答（生产者和kafka集群交互）

ack=0：生产者发送过来的数据，不需要等数据落盘应答（生产者发送过来数据就不管了，可靠性差，效率高；生产中很少使用）

ack=1：生产者发送过来的数据，Leader收到数据后应答（可靠性中等，效率中等；一般用于传输普通日志，允许丢个别数据；）

ack=-1：

生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答（可靠性高，效率低；一般用于传输和钱相关的数据，对可靠性要求比较高的场景）

Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader集合(leader：0，isr:0,1,2)。

如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。例如2超时，(leader:0, isr:0,1)。这样就不用等长期联系不上或者已经故障的节点

我们现在保证了数据尽量不丢失的传输过来，那么怎么最大程度保证数据去重呢？

数据去重

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。

精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2） 

重复数据的判断标准：具有相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。

所以幂等性只能保证的是在单分区单会话内不重复。

如何使用幂等性？

开启参数 enable.idempotence 默认为 true，false 关闭

kafka高效原理

1. 分布式架构：Kafka采用了分布式架构，将数据分散存储在多个服务器节点上，每个节点负责一部分数据的读写操作。这样可以实现数据的并行处理，提高整体的吞吐量和性能。

2. 零拷贝机制：Kafka使用零拷贝技术，在数据读写过程中避免了不必要的数据拷贝操作。当数据从磁盘读取或者写入到磁盘时，Kafka能够直接将数据从内核缓冲区读取或写入，而不需要经过用户空间的拷贝，减少了CPU和内存的开销。

3. 批量处理：Kafka支持批量处理消息，即将多个消息一次性发送或者消费。通过将消息进行批量打包，可以减少网络传输的次数，提高数据的传输效率。同时，批量处理还可以减少磁盘IO和文件系统的开销，提高数据的写入和读取速度。

4. 顺序写磁盘：Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

理论太多了，看起来云里雾里的，要是第一次接触的话估计头都是懵懵的。这里建议最好建个环境实地感受一下，自己摸索一下就算是输入命令行观察一下输出的变化也好，否则没有正反馈​坚持不下来。

这边参考的尚硅谷的高清pdf课件比较体系化，B站和尚硅谷官网都有课件和视频，感兴趣的可以去看看~​