Kafka简单入门

1 介绍Kafka

1.1 什么是事件流式处理？

从技术上讲，“事件流”是以事件 流的形式从数据库、传感器、移动设备、云服务和软件应用程序等事件产生源实时捕获数据的做法。

• 持久化存储这些数据流，可以提供给以后进行检索
• 实时和回顾性地操作、处理和响应事件流
• 并根据需要将数据流路由到不同的平台或者处理程序中

所以事件流确保了数据的连续流动和解释，以便在正确的时间将正确的信息放在正确的位置

流式处理平台的特性

• 可以让你发布和订阅流式的记录，这一方面与消息队列或者企业消息系统类似
• 可以存储流式的记录，并且有较好的容错性
• 可以在流式记录产生时就进行处理Kafka结合了三个关键功能
• 发布（写入）和订阅（读取）事件流，包括从其他系统连续导入/导出数据
• 可以可靠的持久化事件流
• 能够处理实时产生的事件流或者回溯处理历史的事件流

并且这些功能都可以以分布式、高度可扩展、弹性、容错和安全的方式提供，且可以部署在裸机硬、虚拟机和容器上，也可以部署在本地和云中，自行选择

1.2 Kafka的一些概念和术语

• Kafka是一个分布式系统，服务端和客户端通过高性能TCP网络协议进行通信 - 服务端：作为一个或多个服务器集群运行，具有高度可扩展性和容错性：能够在发生故障时，其他服务器接管工作保证继续运行且不丢失数据- 客户端：通过编写分布式应用和微服务，在网络问题或计算机故障时也能以容错的方式并行、大规模读取、写入和处理事件流，且提供了多语言客户端，java、Scala等
• Kafka作为一个集群，运行在一台或者多台服务器上.
• Kafka 通过 topic 对存储的流数据进行分类，且消费者和生产者是完全解耦的——高扩展性的关键因素
• 一个topic可以有0——多个生产和消费者
• topic存在分区概念-partition，且分区可以有副本分布在不同的kafka节点中 - 为了数据的容错性和高可用性，可以复制每个主题，甚至可以跨地理区域或数据中心进行复制，以防出现问题后没有数据副本导致数据丢失- 一般的复制因子为3，始终有3个数据副本，且复制是在topic-partition基本执行的
• 每条记录中包含一个key，一个value和一个timestamp（时间戳）。

1.3 Kafka的四个核心API

• The Producer API 允许一个应用程序发布一串流式的数据到一个或者多个Kafka topic。
• The Consumer API 允许一个应用程序订阅一个或多个 topic ，并且对发布给他们的流式数据进行处理。
• The Streams API 允许一个应用程序作为一个_流处理器_，消费一个或者多个topic产生的输入流，然后生产一个输出流到一个或多个topic中去，在输入输出流中进行有效的转换。
• The Connector API 允许构建并运行可重用的生产者或者消费者，将Kafka topics连接到已存在的应用程序或者数据系统。比如，连接到一个关系型数据库，捕捉表（table）的所有变更内容。

1.4 Topic、日志和消费者

topic和其日志

kafka的核心概念就是：提供一串流式的记录-topic

• topic就是数据主题，是数据记录发布的地方，不同的topic可以区分业务系统
• 每个topic的partition都有一个日志文件，对应的消息会存入其中，且默认存在两天
• offset由消费者控制，通常以线性的方式增加offset
• partition可以通过扩展从而达到扩大单个topic的容量，可以作为并行的单元集被同时消费
• 每个partition都有一个leader，处理partition的读写，对其他的follwers同步消费者
• 每个实例都是分区唯一的消费者。维护消费组中的消费关系由Kafka协议动态处理。 - 如果新的实例加入组，他们将从组中其他成员处接管一些 partition 分区- 如果一个实例消失，拥有的分区将被分发到剩余的实例。
• Kafka 只保证分区内的记录是有序的，而不保证主题中不同分区的顺序。 - 每个 partition 分区按照key值排序足以满足大多数应用程序的需求。
• Kafka保证发送到特定主题和分区的消息，如果越先发送，那么在日志中的偏移值就越低

1.5 Kafka的能力

消息系统能力
相比传统的消息系统，Kafka的消息被读取后还可以被保存在日志中，以便新的消费者从头消费，且可以广播给多个消费组
Kafka有更严格的顺序保证，因为传统消息系统，消息会被异步的传递，可能无序到达导致无序消费
Kafka可以让一个分区对应一个消费者来并行消费，且提供顺序保证和负载均衡
持久化能力
kafka作为消息的中间数据存储系统，能够通过备份保证容错，也就是消息到来后可以通过完全写入和写入副本后才被认为完成写入
使用磁盘结构，50kb和50TB的数据表现都是一致的，通过offset来控制读取位置，是一种高性能、低延迟、具备日志存储、备份和传播的分布式文件系统
流处理能力
Kafka最终目的是为了能够实时的流处理，可以将整个kafka看作是一个stream管道，不同的topic就是不同的节点，对stream中的数据流做不同的操作，对java的stream流也是这种概念
Kafka提供了Stream API去对一些复杂的数据变换，比如流数据的聚合或join，如果是简单的操作，那就用消费者呗
Stream API建立在kafka核心之上，使用消费者和生产者的API作为输入，在kafka中流转，且利用消费组的一些机制保证容错

2 使用的一些案例

消息传递

Kafka可以替代传统的消息代理，因为有更好的吞吐量、内置分区、复制、容错、持久化能力

活动跟踪

通过用户的页面访问活动，将不同的行为发布到不同的中心主题，可以被实时处理，实时监控，以及Hadoop或者离线数仓进行离线分析

指标

kafka可以通过各个topic的数据聚合或者数据处理后的结果产生监控数据，实时监控各项指标

日志聚合

kafka可以替代日志聚合解决方案的替代品，kafka不需要从服务器收集物理日志并放在中心位置处理，而是抽象了文件的详细信息，将日志或事件数据作为消息流进行实时低延迟的处理，且支持多个数据源和多个消费者统一处理

流处理

kafka可以将多个主题的管道 组成 “流处理的步骤管道”，原始输入数据从kafka主题中消费，聚合、转换为其他主题进一步处理的流。

事件溯源

由于kafka的offset和分区以及消息的时序性有保证，所以可以通过这个特性满足一系列操作的时间线，从而去完成各个事件的溯源

提交日志

Kafka 可以充当分布式系统的一种外部提交日志。日志有助于在节点之间复制数据，并充当故障节点恢复数据的重新同步机制，日志压缩功能有助于这种行为

3 快速入门

第一步:获取kafka

下载 最新的 Kafka 版本并解压：

$ tar-xzf kafka_2.13-3.5.0.tgz
$ cd kafka_2.13-3.5.0

第二步:启动环境

Kafka 可以使用 ZooKeeper 或 KRaft 启动

Kafka和ZooKeeper的环境
启动zookeeper

# Start the ZooKeeper service 启动zookeeper
$ bin/zookeeper-server-start.sh config/zookeeper.properties

在另一个终端启动kafka

# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties

Kafka和Kraft的环境
生成集群 UUID

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

设置日志目录格式

$ bin/kafka-storage.sh format-t$KAFKA_CLUSTER_ID-c config/kraft/server.properties

启动Kafka服务器

$ bin/kafka-server-start.sh config/kraft/server.properties

第三步:创建主题存储事件

如果想要发送一个事件或者消息，必须要在主题的基础上，也就是必须要向一个主题发送消息和事件

创建topic名称为quickstart-events，指定本机的9092端口的kafka

$ bin/kafka-topics.sh --create--topic quickstart-events --bootstrap-server localhost:9092

显示主题的分区计数等详细信息

$ bin/kafka-topics.sh --describe--topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events        TopicId: NPmZHyhbR9y00wMglMH2sg PartitionCount: 1       ReplicationFactor: 1    Configs:
    Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

第四步:将事件写入主题(生产)

Kafka 客户端通过网络与 Kafka 代理进行通信以写入（或读取）事件，一旦收到，代理将以持久且容错的方式存储事件，只要您需要，甚至永远存储。

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event

一般来说开启producer后，每一行都是一个单独的事件，CTRL-C可以推出

第五步:阅读事件(消费)

打开另一个终端会话并运行控制台消费者客户端来读取刚刚创建的事件

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

Ctrl-C

可以随时停止消费者客户端。