官网
Apache Kafka是一个开源的分布式事件流平台，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。
核心能力
- 高吞吐量
在网络有限的吞吐量下，使用延迟低至2ms的机器集群交付消息。 - 可扩展性
将生产集群扩展到1000个代理，每天数万亿的消息，PB(petabytes)的数据，数十万个分区。弹性地扩展和收缩存储和处理。
1PB = 1024TB = 2^50字节 - 永久储存
将数据流安全地存储在分布式、持久、容错的集群中。 - 高可用性
在可用性区域上有效地扩展集群，或者跨地理区域连接单独的集群。
生态系统
-
内置流处理
使用事件时间和只处理一次的处理，通过连接、聚合、过滤器、转换等处理事件流。 -
几乎可以连接任何东西
Kafka的开箱即用Connect接口集成了数百个事件源和事件接收，包括Postgres、JMS、Elasticsearch、AWS S3等。 -
CLIENT LIBRARIES
在大量编程语言中读取、写入和处理事件流。 -
大型生态系统开源工具
开放源码工具的大型生态系统:利用大量社区驱动的工具。
信任和易用性
-
关键任务
支持任务关键型用例，保证排序、零消息丢失和高效的只处理一次。 -
被成千上万的组织所信任
成千上万的组织使用Kafka，从互联网巨头到汽车制造商再到证券交易所。超过500万的独立终身下载量。 -
庞大的用户群
Kafka是Apache软件基金会五个最活跃的项目之一，在世界各地有数百个聚会。 -
丰富的在线资源
丰富的文档，在线培训，指导教程，视频，示例项目，Stack Overflow等。
1、简介
1.1 什么是事件流?
事件流(Event streaming)是相当于人体中枢神经系统的数字系统。它是“永远在线(always-on)”世界的技术基础，在这个世界中，业务越来越多地由软件定义和自动化，软件的用户更多地是软件。
从技术上讲，事件流是以事件流(streams of events)的形式从数据库、传感器、移动设备、云服务和软件应用程序等事件源实时捕获数据的实践；持久地存储这些事件流以供以后检索；实时和回顾性地操作、处理和响应事件流；并根据需要将事件流路由到不同的目标技术。因此，事件流确保了数据连续的流动和解释，从而使正确的信息在正确的时间出现在正确的位置。
1.2 我可以使用事件流做什么?
事件流被应用到大量行业和组织中的各种用例中。它的许多例子包括:
- 实时处理支付和金融交易，如在证券交易所、银行和保险公司。
- 实时跟踪和监控汽车、卡车、车队和货物，例如在物流和汽车行业。
- 持续捕获和分析来自物联网设备或其他设备(如工厂和风力发电场)的传感器数据。
- 收集并立即响应客户交互和订单，例如在零售、酒店和旅游行业以及移动应用程序中。
- 监测病人在医院的护理和预测病情的变化，以确保在紧急情况下及时治疗。
- 连接、存储和提供由公司不同部门产生的数据。
- 作为数据平台、事件驱动架构和微服务的基础。
1.3 Apache Kafka®是一个事件流平台。这意味着什么?
Kafka结合了三个关键功能，因此您可以通过一个经过实战测试的解决方案实现端到端的事件流用例:
- 发布(写)
publish (write)和订阅(读)subscribe to (read)事件流，包括从其他系统连续导入/导出数据。 - 持久可靠地存储事件流，只要你想。
- 当事件发生时或回顾性地处理事件流
所有这些功能都以分布式、高度可伸缩、弹性、容错和安全的方式提供。Kafka可以部署在裸机硬件、虚拟机和容器上，也可以部署在本地和云中。你可以在自我管理Kafka环境和使用各种供应商提供的完全托管服务之间进行选择。
1.4 简而言之，Kafka是如何工作的?
Kafka是一个分布式系统，由服务器（Server）和客户端(client)组成，通过高性能TCP网络协议进行通信。它可以部署在本地和云环境中的裸机硬件、虚拟机和容器上。
服务器：Kafka作为一个或多个服务器的集群运行，可以跨越多个数据中心或云区域。其中一些服务器构成存储层，称为代理(broker)。其他服务器运行Kafka Connect，以事件流的形式持续导入和导出数据，将Kafka与现有系统(如关系数据库以及其他Kafka集群)集成。为了让你实现关键任务用例，Kafka集群具有高度可扩展性和容错性:如果其中任何一个服务器出现故障，其他服务器将接管它们的工作，以确保持续运行而不会丢失任何数据。
客户端 ：它们允许您编写分布式应用程序和微服务，这些应用程序和微服务可以并行地、大规模地、以容错的方式读取、写入和处理事件流，即使在网络问题或机器故障的情况下也是如此。Kafka附带了一些这样的客户端，这些客户端是由Kafka社区提供的数十个客户端增强的:客户端可用于Java和Scala，包括更高级别的Kafka Streams库，用于Go, Python, C/C++和许多其他编程语言以及REST api。
1.5 主要概念和术语
事件(event)记录了世界上或你的业务中“发生了什么事”的事实。在文档中也称为记录(record)或消息(message)。当你向Kafka读写数据时，你是以事件的形式来完成的。从概念上讲，事件有键(key)、值(value)、时间戳(timestamp)和可选的元数据头(metadata headers)。下面是一个例子:
- Event key: “Alice”
- Event value: “Made a payment of $200 to Bob”
- Event timestamp: “Jun. 25, 2020 at 2:06 p.m.”
生产者(Producer)是那些向Kafka发布(写入)事件的客户端应用程序，消费者(consumer)是那些订阅(读取和处理)这些事件的客户端应用程序。在Kafka中，生产者和消费者是完全解耦的，彼此不可知，这是实现Kafka所知的高可扩展性的关键设计元素。例如，生产者永远不需要等待消费者。Kafka提供了各种保证，比如只处理一次事件的能力。
事件被组织并持久地存储在主题(topic)中。非常简单地说，主题类似于文件系统中的一个文件夹，而事件就是该文件夹中的文件。例如，主题名称可以是“payments”。Kafka中的主题总是多生产者和多订阅者:一个主题可以有零、一个或多个生产者向其写入事件，也可以有零、一个或多个消费者订阅这些事件。主题中的事件可以根据需要随时读取——与传统的消息传递系统不同，事件在使用后不会被删除。相反，你可以通过每个主题的配置设置来定义Kafka应该保留多长时间的事件，之后旧的事件将被丢弃。Kafka的性能在数据大小方面是有效的常数，所以长时间存储数据是完全没问题的。
主题是分区的(partitioned)，这意味着一个主题分布在位于不同Kafka代理上的许多“桶(buckets)”上。数据的这种分布式放置对于可伸缩性非常重要，因为它允许客户机应用程序同时从多个代理读取和写入数据。当新事件发布到主题时，它实际上被附加到主题的一个分区中。具有相同事件键(例如，客户或车辆ID)的事件被写入相同的分区，并且Kafka保证给定主题分区的任何消费者总是以与写入相同的顺序读取该分区的事件。
图:这个示例主题有四个分区P1-P4。两个不同的生产者客户端通过网络将事件写入主题的分区，彼此独立地向主题发布新事件。具有相同键(在图中用颜色表示)的事件被写入相同的分区。注意，如果合适的话，两个生产者都可以写入同一个分区。
为了使您的数据具有容错性和高可用性，可以复制(replicated)每个主题，甚至跨地理区域或数据中心，以便始终有多个拥有数据副本的代理，以防出现问题，您需要对代理进行维护，等等。一个常见的生产设置是复制因子为3，也就是说，您的数据将始终有三个副本。这种复制是在主题分区级别执行的。
对于入门来说，这篇初级文章应该足够了。文档的设计部分详细解释了Kafka的各种概念，如果你感兴趣的话。
1.6 Kafka APIs
除了用于操作和管理任务的命令行工具外，Kafka还为Java和Scala提供了五个核心api:
- Admin API用于管理和检查主题、代理和其他Kafka对象。
- Producer API 发布(写)一个事件流到一个或多个Kafka主题。
- Consumer API 订阅(阅读)一个或多个主题，并处理产生给这些主题的事件流。
- Kafka Streams API 实现流处理应用和微服务。它提供了更高级的功能来处理事件流，包括转换、有状态操作(如聚合和连接)、窗口、基于事件时间的处理等等。从一个或多个主题读取输入，以便为一个或多个主题生成输出，从而有效地将输入流转换为输出流。
- Kafka Connect API 构建和运行可重用的数据导入/导出连接器，这些连接器可以从外部系统和应用程序中消费(读取)或产生(写入)事件流，以便与Kafka集成。例如，连接到关系数据库(如PostgreSQL)的连接器可能会捕获对一组表的每一次更改。然而，在实践中，你通常不需要实现自己的连接器，因为Kafka社区已经提供了数百个现成的连接器。
1.7 接下来该怎么办
- 要获得Kafka的实践经验，请遵循快速入门。
- 要更详细地了解Kafka，请阅读文档。你也可以选择Kafka 的书籍和学术论文。
- 浏览用例，了解我们全球社区的其他用户如何从Kafka中获得价值。
- 加入当地的Kafka聚会小组，观看Kafka峰会(Kafka社区的主要会议)的演讲。
2、quickstart
步骤1: 获取kafka
下载最新的Kafka版本并解压:
$ tar -xzf kafka_2.13-3.4.0.tgz
$ cd kafka_2.13-3.4.0
步骤2: 启动kafka环境
注意:您的本地环境必须安装Java 8+
Apache Kafka可以使用ZooKeeper或KRaft启动。要开始使用任何一种配置，请按照下面的一节进行操作，但不要同时进行。
用ZooKeeper
运行如下命令，以正确的顺序启动所有服务:
# Start the ZooKeeper service
$ bin/zookeeper-server-start.sh config/zookeeper.properties
打开另一个终端会话并运行:
# 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
一旦Kafka服务器成功启动，你将有一个基本的Kafka环境运行并准备使用。
步骤3: 创建一个主题来存储事件
Kafka是一个分布式事件流平台，允许您跨多个机器读取，写入，存储和处理事件(也称为文档中的记录或消息)。
事件示例包括支付交易、移动电话的地理位置更新、发货订单、物联网设备或医疗设备的传感器测量等等。这些事件被组织并存储在主题中。非常简单地说，主题类似于文件系统中的一个文件夹，而事件就是该文件夹中的文件。
因此，在您编写第一个事件之前，您必须创建一个主题。打开另一个终端会话并运行:
$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
# ./bin/kafka-topics.sh --help
This tool helps to create, delete, describe, or change a topic.
Option Description
------ -----------
--alter Alter the number of partitions,
replica assignment, and/or
configuration for the topic.
--at-min-isr-partitions if set when describing topics, only
show partitions whose isr count is
equal to the configured minimum.
--bootstrap-server <String: server to REQUIRED: The Kafka server to connect
connect to> to.
#要连接的Kafka服务器
--command-config <String: command Property file containing configs to be
config property file> passed to Admin Client. This is used
only with --bootstrap-server option
for describing and altering broker
configs.
--config <String: name=value> A topic configuration override for the
topic being created or altered. The
following is a list of valid
configurations:
cleanup.policy
compression.type
delete.retention.ms
file.delete.delay.ms
flush.messages
flush.ms
follower.replication.throttled.
replicas
index.interval.bytes
leader.replication.throttled.replicas
local.retention.bytes
local.retention.ms
max.compaction.lag.ms
max.message.bytes
message.downconversion.enable
message.format.version
message.timestamp.difference.max.ms
message.timestamp.type
min.cleanable.dirty.ratio
min.compaction.lag.ms
min.insync.replicas
preallocate
remote.storage.enable
retention.bytes
retention.ms
segment.bytes
segment.index.bytes
segment.jitter.ms
segment.ms
unclean.leader.election.enable
See the Kafka documentation for full
details on the topic configs. It is
supported only in combination with --
create if --bootstrap-server option
is used (the kafka-configs CLI
supports altering topic configs with
a --bootstrap-server option).
--create Create a new topic.
--delete Delete a topic
--delete-config <String: name> A topic configuration override to be
removed for an existing topic (see
the list of configurations under the
--config option). Not supported with
the --bootstrap-server option.
--describe List details for the given topics.
#列出给定主题的详细信息。
--disable-rack-aware Disable rack aware replica assignment
--exclude-internal exclude internal topics when running
list or describe command. The
internal topics will be listed by
default
--help Print usage information.
--if-exists if set when altering or deleting or
describing topics, the action will
only execute if the topic exists.
--if-not-exists if set when creating topics, the
action will only execute if the
topic does not already exist.
--list List all available topics.
--partitions <Integer: # of partitions> The number of partitions for the topic
being created or altered (WARNING:
If partitions are increased for a
topic that has a key, the partition
logic or ordering of the messages
will be affected). If not supplied
for create, defaults to the cluster
default.
--replica-assignment <String: A list of manual partition-to-broker
broker_id_for_part1_replica1 : assignments for the topic being
broker_id_for_part1_replica2 , created or altered.
broker_id_for_part2_replica1 :
broker_id_for_part2_replica2 , ...>
--replication-factor <Integer: The replication factor for each
replication factor> partition in the topic being
created. If not supplied, defaults
to the cluster default.
--topic <String: topic> The topic to create, alter, describe
or delete. It also accepts a regular
expression, except for --create
option. Put topic name in double
quotes and use the '\' prefix to
escape regular expression symbols; e.
g. "test\.topic".
#要创建、修改、描述(显示)或删除的主题。除了--create选项之外，它还
#接受正则表达式。将主题名称放在双引号中，并使用'\'前缀转义正
#则表达式符号;如："test\.topic"。
--topic-id <String: topic-id> The topic-id to describe.This is used
only with --bootstrap-server option
for describing topics.
--topics-with-overrides if set when describing topics, only
show topics that have overridden
configs
--unavailable-partitions if set when describing topics, only
show partitions whose leader is not
available
--under-min-isr-partitions if set when describing topics, only
show partitions whose isr count is
less than the configured minimum.
--under-replicated-partitions if set when describing topics, only
show under replicated partitions
--version Display Kafka version.
所有Kafka的命令行工具都有额外的选项(options):运行kafka-topics.sh命令，不带任何参数来显示使用信息。例如，它还可以显示新主题的分区计数等详细信息:
$./bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events TopicId: 4HukHBG9QYySmD7mtOIYbw PartitionCount: 1 ReplicationFactor: 1 Configs:
Topic: quickstart-events Partition: 0 Leader: 0 Replicas: 0 Isr: 0
步骤4: 把一些事件写进主题
Kafka客户端通过网络与Kafka代理通信，以写(或读)事件。一旦接收到事件，代理将以持久和容错的方式存储事件，只要您需要——甚至是永远。
运行控制台生产者客户端，将一些事件写入主题。默认情况下，您输入的每一行都将导致将一个单独的事件写入主题。
$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event
您可以随时使用Ctrl-C停止生产者客户端。
步骤5: 读取事件
打开另一个终端会话并运行console消费者客户端来读取你刚刚创建的事件:
$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event
# ./bin/kafka-console-consumer.sh --help
This tool helps to read data from Kafka topics and outputs it to standard output.
Option Description
------ -----------
--bootstrap-server <String: server to REQUIRED: The server(s) to connect to.
connect to>
--consumer-property <String: A mechanism to pass user-defined
consumer_prop> properties in the form key=value to
the consumer.
--consumer.config <String: config file> Consumer config properties file. Note
that [consumer-property] takes
precedence over this config.
--enable-systest-events Log lifecycle events of the consumer
in addition to logging consumed
messages. (This is specific for
system tests.)
--formatter <String: class> The name of a class to use for
formatting kafka messages for
display. (default: kafka.tools.
DefaultMessageFormatter)
--formatter-config <String: config Config properties file to initialize
file> the message formatter. Note that
[property] takes precedence over
this config.
--from-beginning If the consumer does not already have
an established offset to consume
from, start with the earliest
message present in the log rather
than the latest message.
# 如果消费者还没有一个确定的消费偏移量，那么从日志中出现的最
# 早的消息开始，而不是从最新的消息开始。
--group <String: consumer group id> The consumer group id of the consumer.
--help Print usage information.
--include <String: Java regex (String)> Regular expression specifying list of
topics to include for consumption.
--isolation-level <String> Set to read_committed in order to
filter out transactional messages
which are not committed. Set to
read_uncommitted to read all
messages. (default: read_uncommitted)
--key-deserializer <String:
deserializer for key>
--max-messages <Integer: num_messages> The maximum number of messages to
consume before exiting. If not set,
consumption is continual.
--offset <String: consume offset> The offset to consume from (a non-
negative number), or 'earliest'
which means from beginning, or
'latest' which means from end
(default: latest)
--partition <Integer: partition> The partition to consume from.
Consumption starts from the end of
the partition unless '--offset' is
specified.
--property <String: prop> The properties to initialize the
message formatter. Default
properties include:
print.timestamp=true|false
print.key=true|false
print.offset=true|false
print.partition=true|false
print.headers=true|false
print.value=true|false
key.separator=<key.separator>
line.separator=<line.separator>
headers.separator=<line.separator>
null.literal=<null.literal>
key.deserializer=<key.deserializer>
value.deserializer=<value.
deserializer>
header.deserializer=<header.
deserializer>
Users can also pass in customized
properties for their formatter; more
specifically, users can pass in
properties keyed with 'key.
deserializer.', 'value.
deserializer.' and 'headers.
deserializer.' prefixes to configure
their deserializers.
--skip-message-on-error If there is an error when processing a
message, skip it instead of halt.
--timeout-ms <Integer: timeout_ms> If specified, exit if no message is
available for consumption for the
specified interval.
--topic <String: topic> The topic to consume on.
--value-deserializer <String:
deserializer for values>
--version Display Kafka version.
--whitelist <String: Java regex DEPRECATED, use --include instead;
(String)> ignored if --include specified.
Regular expression specifying list
of topics to include for consumption.
您可以随时使用Ctrl-C停止消费者客户端。
请随意尝试:例如，切换回生产者终端(上一步)来编写额外的事件，并查看事件如何立即显示在消费者终端中。
因为事件被持久地存储在Kafka中，它们可以被你想要的尽可能多的消费者读取多次。您可以很容易地通过打开另一个终端会话并再次重新运行前面的命令来验证这一点。
步骤6: 使用kafka connect导入/导出数据作为事件流
您可能在现有系统(如关系数据库或传统消息传递系统)中有大量数据，以及许多已经使用这些系统的应用程序。Kafka Connect允许你不断地从外部系统摄取数据到Kafka，反之亦然。它是一个可扩展的工具，运行连接器，实现用于与外部系统交互的自定义逻辑。因此，将现有系统与Kafka集成非常容易。为了使这个过程更容易，有数百个这样的连接器随时可用。
在这个快速入门中，我们将看到如何使用简单的连接器来运行Kafka Connect，将数据从文件导入到Kafka主题，并将数据从Kafka主题导出到文件。
首先，确保将connect-file-3.4.0.jar添加到插件中。在Connect worker的配置中设置plugin.path。为了快速入门，我们将使用一个相对路径，并将连接器的包视为一个uber jar，当快速入门命令从安装目录运行时，它就会工作。然而，值得注意的是，对于生产部署，使用绝对路径总是可取的。看到plugin.path，用于获取如何设置此配置的详细说明。
编辑config/connect-standalone.properties文件，添加或更改plugin.path配置属性匹配如下，并保存文件:
echo "plugin.path=libs/connect-file-3.4.0.jar"
然后，首先创建一些种子数据进行测试:
echo -e "foo\nbar" > test.txt
Or on Windows:
echo foo> test.txt
echo bar>> test.txt
接下来，我们将启动以独立模式运行的两个连接器，这意味着它们在单个本地专用进程中运行。我们提供了三个配置文件作为参数。第一个总是Kafka Connect进程的配置，包含常见的配置，比如Kafka要连接的代理和数据的序列化格式。其余的配置文件分别指定要创建的连接器。这些文件包括唯一的连接器名称、要实例化的连接器类以及连接器所需的任何其他配置。
bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties
这些示例配置文件，包含在Kafka中，使用您之前启动的默认本地集群配置并创建两个连接器:第一个是源连接器，它从输入文件中读取行并将每个行生成到Kafka主题，第二个是接收器连接器，它从Kafka主题中读取消息并将每个消息作为输出文件中的一行生成。
在启动期间，您将看到许多日志消息，包括一些指示连接器正在实例化的消息。一旦Kafka Connect进程启动，源连接器应该开始从test.txt中读取行并将其生成到主题connect-test，接收器连接器应该开始从主题connect-test中读取消息并将其写入文件test.sink.txt。我们可以通过检查输出文件的内容来验证数据已经通过整个管道传递:
> more test.sink.txt
foo
bar
请注意，数据被存储在Kafka主题connect-test中，所以我们也可以运行一个控制台消费者来查看主题中的数据(或者使用自定义消费者代码来处理它):
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
...
连接器继续处理数据，因此我们可以将数据添加到文件中，并看到它在管道中移动:
echo Another line>> test.txt
步骤7: 用kafka流处理事件
一旦你的数据作为事件存储在Kafka中，你就可以使用为Java/Scala提供的Kafka Streams客户端库处理数据。它允许你实现关键任务的实时应用和微服务，其中输入和/或输出数据存储在Kafka主题中。Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性，以及Kafka服务器端集群技术的优势，使这些应用程序具有高度可扩展性、弹性、容错性和分布式。该库支持一次处理、有状态操作和聚合、窗口、连接、基于事件时间的处理等等。
为了给你一个初步的体验，下面是如何实现流行的WordCount算法:
KStream<String, String> textLines = builder.stream("quickstart-events");
KTable<String, Long> wordCounts = textLines
.flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
.groupBy((keyIgnored, word) -> word)
.count();
wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));
Kafka Streams演示和应用程序开发教程演示了如何从头到尾编写和运行这样一个流应用程序。
步骤8: 终止kafka环境
现在您已经完成了快速入门，可以随意终止Kafka环境，或者继续使用。
- 使用
Ctrl-C停止生产者和消费者客户端(如果您还没有这样做的话)。 - 用
Ctrl-C停止Kafka代理。 - 最后，如果Kafka with ZooKeeper部分被启动，用
Ctrl-C停止ZooKeeper服务器。
如果你还想删除本地Kafka环境的任何数据，包括你在此过程中创建的任何事件，运行命令:
rm -rf /tmp/kafka-logs /tmp/zookeeper /tmp/kraft-combined-logs
