kafka使用教程、快速上手

kafka概述

• 一、kafka概述
• • 1.1 定义
  • 1.2 消息队列
  • • 1.2.1 传统消息队列的应用场景
    • 1.2.2 消息队列的两种形式
  • 1.3 Kafka 基础架构
• 二、kafka安装部署
• • 2.1安装部署
  • • 2.1.1.jar包下载
    • 2.1.2.解压到指定的文件夹下
    • 2.1.3.创建两个文件夹以供后续使用
    • 2.1.4. 修改配置文件
    • • （1）修改zookeeper.properties 文件
      • （2） 修改server.properties 文件
  • 2.2启动
  • • 2.2.1.启动 kafka 内置的 zookeeper
    • 2.2.2.启动 kafka 服务
    • 2.2.3.创建一个名为 test1 的 topic 测试主题 kafka
    • 2.2.4.创建消息生产者生产消息
    • 2.2.5.创建消息消费者接收消息
    • 2.2.6.测试消息发送和接收
• 三、kafka架构深入理解
• • 3.1 Kafka 工作流程
  • • 3.1.1 写入方式
    • 3.1.2 分区（Partition）
    • • 1.启动zookeeper
      • 2 启动服务
      • 3 创建主题
      • 4 查看主题
      • 5 查看主题列表
  • 3.2 Kafka 存储机制
  • • 3.2.1 数据分片
    • 3.2.2 log分段：
    • 3.2.3 日志的清除策略以及压缩策略
  • 3.3 Kafka 生产者
  • • 3.3.1 数据可靠性保证
  • 3.4 Kafka 消费者
  • • 3.4.1概念
    • 3.4.2 消费方式
    • • 1.消费位移确认
      • 2 以时间戳查询消息
      • 3 消费速度控制
  • 3.5 Kafka 高效读取数据
• 四、kafka API
• • 4.1 Producer API
  • • 4.1.1 消息发送流程
    • 4.1.2 异步发送 API
    • 4.1.3 同步发送 API
  • 4.2 Consumer API
  • • 4.2.1 自动提交 offset
    • 4.2.2 手动提交 offset
    • 4.2.3 自定义存储 offset
  • 4.3 自定义拦截器
  • • 4.3.1 拦截器原理
    • 4.3.2 拦截器案例
• 五、kafka事务
• • 5.1 事务场景
  • 5.2 几个关键概念和推导
  • 5.3 事务语义
  • • 5.3.1 多分区原子写入
    • 5.3.2 粉碎“僵尸实例”
    • 5.3.3 读事务消息
    • 5.4.4 事务处理Java API
    • • 5.4.4.1 api分类
      • 5.4.4.2 事务配置
      • 5.4.4.3 “只有写”应用程序示例
      • 5.4.4.4 消费-生产并存（consume-Transform-Produce）
  • 5.5 事务工作原理
  • • 5.5.1 事务协调器和事务日志
    • 5.5.2 事务数据流
  • 5.6 事务相关配置
  • • 5.6.1 Broker configs
    • 5.6.2 Producer configs
    • 5.6.3 Consumer configs
  • 5.7 事务性能以及如何优化
  • • 5.7.1 Producer打开事务之后的性能
    • 5.7.2 Consumer打开之后的性能
• 六、SpringBoot集成kafka
• • • 6.1 配置Maven依赖
    • 6.2 项目具体代码
    • • 6.2.1 yml配置
      • 6.2.2 生产者
      • 6.2.3 消费者
      • 6.2.4

一、kafka概述

1.1 定义

Kafka 是一个分布式的基于发布 / 订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。

1.2 消息队列

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

使用消息队列的好处：

1. 解耦

   允许独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

2. 可恢复性

   系统的一部分组件失效时，不会影响整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

3. 缓冲

   有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

4. 灵活性和峰值处理能力

   使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

5. 异步通信

   很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

1.2.2 消息队列的两种形式

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

   消息生产者生产消息发送到 Queue 中，然后消费者从 Queue 中取出并且消费消息。消息被消费以后，Queue 中不再有存储，所以消费者不可能消费到已经被消费的消息。Queue 支持存在多个消费者，但对于一个消息而言，只有一个消费者可以消费。

   

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

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

   

1.3 Kafka 基础架构

1. Producer：

   消息生产者，就是向 Kafka broker 发消息的客户端。

2. Consumer：

   消息消费者，向 Kafka broker 取消息的客户端。

3. Consumer Group（CG）：

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

4. Broker：

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

5. Topic：

   可以理解为一个队列，生产者和消费者面向的都是一个 topic。

6. Partiton：

   为了实现拓展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 Partition，每个 partition 都是一个有序的队列。

7. Replication：

   副本，为保证集群中某个节点发生故障时，该节点上的 partition 数据不丢失，且 Kafka 仍然可以继续工作，Kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 leader 和若干个 follower。

8. leader：

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

9. follower：

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

二、kafka安装部署

2.1安装部署

2.1.1.jar包下载

网址：http://kafka.apache.org/downloads.html

!不要下载最新版，在win10下有问题，下载2.8.1如下图

下载Binary版本：

[这里是图片005]

2.1.2.解压到指定的文件夹下

2.1.3.创建两个文件夹以供后续使用

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aiBYooS2-1644458525646)(https://images1.tqwba.com/20201029/5trqdowih4q.png)]

2.1.4. 修改配置文件

（1）修改zookeeper.properties 文件

修改 kafka_2.12-2.8.1configzookeeper.properties 文件 大概第16行

注意文件分隔符是\

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-WL7WD6Hz-1644458525651)(https://images1.tqwba.com/20201029/twfv43l23jb.png)]

（2） 修改server.properties 文件

修改 kafka_2.12-2.8.1configserver.properties 文件 大概第60行

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ql4BDbuW-1644458525655)(https://images1.tqwba.com/20201029/xlzkbjusutq.png)]

2.2启动

2.2.1.启动 kafka 内置的 zookeeper

运行 cmd 命令：

！如果报错 The input line is too long，将文件路径缩小即可，如直接放在C盘下

！如果报错Unable to access datadir，请把修改配置文件时的两个路径均修改为相对路径

.inwindowszookeeper-server-start.bat .configzookeeper.properties
1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rOwCHlMa-1644458525657)(https://images1.tqwba.com/20201029/iksbqbox45t.png)]

不关闭当前窗口

2.2.2.启动 kafka 服务

运行 cmd 命令：

.inwindowskafka-server-start.bat .configserver.properties
1

不关闭当前窗口

2.2.3.创建一个名为 test1 的 topic 测试主题 kafka

运行 cmd 命令：

.inwindowskafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1
1

该窗口可关闭

2.2.4.创建消息生产者生产消息

运行 cmd 命令：

.inwindowskafka-console-producer.bat --broker-list localhost:9092 --topic test1
1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0kvDgrbJ-1644458525662)(https://images1.tqwba.com/20201029/vnifjx1ckrw.png)]

不关闭当前窗口

2.2.5.创建消息消费者接收消息

运行 cmd 命令：

.inwindowskafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test1 --from-beginning
1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lyzsgoOg-1644458525663)(https://images1.tqwba.com/20201029/tjyef1mgnsh.png)]

不关闭当前窗口

2.2.6.测试消息发送和接收