Apache Kafka是一个分布式流媒体平台，流媒体平台有三个关键功能：

1、它允许您发布和订阅记录流。在这方面，它类似于消​​息队列或企业消息系统。

2、它允许您以容错方式存储记录流。

3、它可以让您处理记录流。

Apache Kafka如何做这些事情:

1、Kafka作为一个或多个服务器上的集群运行。
2、Kafka群集将名为主题的类别的记录流存储起来。
3、每个记录由一个键，一个值和一个时间戳组成。

它被用于两大类应用程序：
1、构建可在系统或应用程序之间可靠获取数据的实时流数据流水线
2、构建对数据流进行变换或反应的实时流应用程序

1、Producer API：允许应用程序发布的记录流至一个或多个 Kafka topics
2、Consumer API：允许应用程序订阅一个或多个Kafka topics，并处理所产生的对他们记录的数据流。
3、 Streams API：允许应用程序充当流处理器，从一个或多个Kafka topics消耗的输入流，并产生一个输出流至一个或多个输出的Kafka topics，有效地变换所述输入流，以输出流。

4、Connector API：允许构建和运行将Kafka topics与现有应用程序或数据系统相连接的可重复使用的producers 或consumers 。例如，关系数据库的连接器可能会捕获到表的每个更改。

在Kafka，客户端和服务器之间的通信是用简单，高性能，语言不可知的TCP协议完成的。此协议版本化并保持与旧版本的向后兼容性。我们为Kafka提供Java客户端，但是客户端可以使用多种语言。

我们先来看看kafka提供的一个记录流的核心抽象 - Topics

Topics是发布记录的类别或Feed名称。kafka的Topics总是多用户的; 也就是说，Topics可以有零个，一个或多个消费者订阅数据。
对于每个Topics，Kafka集群维护一个分区日志，如下所示：

每个分区是一个有序的，不可变的记录序列，不断附加到结构化的提交日志中。每个分区中的记录都被分配一个顺序的id号，称为唯一标识分区内每个记录的偏移量（offset）.

Kafka集群保留所有已发布的记录 - 无论它们是否已被使用 - 可以使用可配置的保留期限。例如，如果保留策略设置为两天，则在发布记录后的两天内，它可以消费，之后它将被丢弃以释放空间。Kafka的性能不受数据大小方面的影响，因此长时间存储数据并不成问题。

事实上，在每个消费者基础上保留的唯一元数据是消费者在日志中的偏移或位置。这个偏移由消费者控制：通常消费者会在读取记录时线性地提高其偏移，但实际上，由于位置由消费者控制，它可以以任何顺序消耗记录。例如，消费者可以重置为较旧的偏移量以重新处理来自过去的数据，或者跳过最近的记录，并从“now”开始消费。

这种特征的组合意味着Kafka消费者的成本非常低 - 他们可以来回去对集群或其他消费者没有太大影响。例如，您可以使用我们的命令行工具来“tail”任何主题的内容，而无需更改任何现有消费者所消耗的内容。

日志中的分区有几个目的。首先，它们允许日志扩展到适合单个服务器的大小。每个单独的分区必须适合托管它的服务器，但主题可能有很多分区，因此它可以处理任意数量的数据。第二，它们作为并行性的单位 - 更多的是在这一点上。

日志的分区分布在Kafka集群中的服务器上，每个服务器处理数据并请求分区的一部分。每个分区都跨可配置数量的服务器进行复制，以实现容错。

每个分区有一个服务器，充当“leader”，零个或多个服务器充当“followers”。leader处理分区的所有读取和写入请求，而followers被动地复制leader。如果leader失败，其中一个followers将自动成为新的leader。每个服务器作为其一些分区的leader，并且作为其他分区的followers，因此在集群内负载平衡良好。

生产者将数据发布到他们选择的主题。生产者负责选择要分配给主题中哪个分区的记录。这可以在一个循环的方式进行简单地平衡负载，也可以根据一些语义分区功能（比如基于记录一些关键）来完成。

消费者 用消费者群组名称标注自己，并将发布到主题的每条记录传递到每个订阅消费者组中的一个消费者实例。消费者实例可以在单独的进程中或在单独的机器上。
如果所有的消费者实例具有相同的消费者组，则记录将在消费者实例上有效地负载平衡。
如果所有的消费者实例都有不同的消费者群体，那么每个记录将被广播给所有的消费者进程。

两个服务器Kafka集群托管四个分区（P0-P3）与两个消费者组。消费者组A有两个消费者实例，B组有四个。

在高级kafka提供以下保证：

生产者producer发送到特定topics分区的消息将按照发送的顺序进行追加。也就是说，如果记录M1由与记录M2相同的producer 发送，并且首先发送M1，则M1将具有比M2更低的偏移offsers并且在日志中较早出现。
消费者实例按照它们存储在日志中的顺序查看记录。
对于具有复制因子N的主题，我们将容忍最多N-1个服务器故障，而不会丢失提交到日志的任何记录。

kafka的流量概念与传统的企业邮件系统相比如何？

消息传统上有两种模式：队列和发布 - 订阅。在队列中，消费者池可以从服务器读取，每条记录都转到其中一个; 在发布订阅中，记录将广播给所有消费者。这两个模型中的每一个都有实力和弱点。排队的优点是它允许您在多个消费者实例上分配数据处理，从而可以扩展您的处理。不幸的是，队列不是多用户的 - 一旦一个进程读取数据就会消失。发布订阅允许您将数据广播到多个进程，但无法缩放处理，因为每个消息都发送给每个用户。

kafka消费群体概念概括了这两个概念。与队列一样，消费者组允许您通过一系列进程（消费者组的成员）来划分处理。与发布订阅一样，Kafka允许您将消息广播到多个消费者组。

kafka模型的优点是，每个主题都具有这两个属性 - 它可以扩展处理，也是多用户 - 不需要选择一个或另一个。

kafka也比传统的邮件系统更强大的订购保证。

传统队列在服务器上保存顺序的记录，如果多个消费者从队列中消费，则服务器按照存储顺序输出记录。然而，虽然服务器按顺序输出记录，但是记录被异步传递给消费者，所以它们可能会在不同的消费者处按顺序到达。这意味着在并行消耗的情况下，记录的排序丢失。消息传递系统通常通过使“唯一消费者”的概念只能让一个进程从队列中消费，但这当然意味着处理中没有并行性。

kafka做得更好 通过在主题中有一个并行概念（分区），Kafka能够在消费者流程池中提供排序保证和负载平衡。这是通过将主题中的分区分配给消费者组中的消费者来实现的，以便每个分区由组中的一个消费者消耗。通过这样做，我们确保消费者是该分区的唯一读者，并按顺序消耗数据。由于有许多分区，这仍然平衡了许多消费者实例的负载。但请注意，消费者组中的消费者实例不能超过分区。

允许发布消息消除消息的消息队列有效地充当飞行中消息的存储系统。Kafka的不同之处在于它是一个很好的存储系统。

写入Kafka的数据写入磁盘并进行复制以进行容错。Kafka允许生产者等待确认，以便在完全复制之前写入不被认为是完整的，并且即使服务器写入失败，也保证持久写入。
Kafka的磁盘结构使用缩放，Kafka将执行相同的操作，无论您在服务器上是否有50 KB或50 TB的持久数据。

作为严重存储并允许客户端控制其读取位置的结果，您可以将Kafka视为专用于高性能，低延迟的提交日志存储，复制和传播的专用分布式文件系统。

仅读取，写入和存储数据流是不够的，目的是实现流的实时处理。

在Kafka，流处理器是从输入主题接收数据流的任何东西，对此输入执行一些处理，并生成持续的数据流以输出主题。

例如，零售应用程序可能会收到销售和出货的输入流，并输出根据该数据计算的重新排序和价格调整。

可以直接使用生产者和消费者API进行简单处理。然而对于更复杂的转换，Kafka提供了一个完全集成的Streams API。这允许构建应用程序进行非平凡处理，以计算流中的聚合或将流连接在一起。

该设施有助于解决这种类型的应用程序面临的困难问题：处理无序数据，重新处理输入作为代码更改，执行有状态计算等。

流API基于Kafka提供的核心原语构建：它使用生产者和消费者API进行输入，使用Kafka进行有状态存储，并在流处理器实例之间使用相同的组机制来实现容错。

消息，存储和流处理的这种组合似乎是不寻常的，但是Kafka作为流媒体平台的角色至关重要。

像HDFS这样的分布式文件系统允许存储用于批处理的静态文件。有效像这样的系统允许存储和处理历史从过去的数据。

传统的企业邮件系统允许处理将在您订阅之后到达的未来邮件。以这种方式构建的应用程序在未来数据到达时处理。

Kafka结合了这两种功能，组合对于Kafka作为流应用程序和流数据管道平台来说至关重要。

通过组合存储和低延迟订阅，流式应用程序可以以相同的方式处理过去和未来的数据。这是一个单一的应用程序可以处理历史的，存储的数据，而不是在到达最后一个记录时结束，它可以在将来的数据到达时继续处理。这是一个广泛的流处理概念，其中包含批处理以及消息驱动应用程序。

同样，对于流数据流水线，订阅到实时事件的组合使得可以使用Kafka进行非常低延迟的管道; 但是可靠性地存储数据的能力使得可以将其用于必须保证数据传送的关键数据，或者与仅负载数据的离线系统集成，或者可能会长时间停机以进行维护。流处理设备可以在数据到达时转换数据。

https://www.apache.org/dyn/closer.cgi?path=/kafka/0.10.2.0/kafka_2.11-0.10.2.0.tgz

tar -xzf kafka_2.11-0.10.2.0.tgz

cd kafka_2.11-0.10.2.0

Kafka使用ZooKeeper，所以您需要先启动一个ZooKeeper服务器，如果您还没有。您可以使用随kafka一起打包的便捷脚本来获取一个快速的单节点ZooKeeper实例。

后台运行加符号&

bin/zookeeper-server-start.sh config/zookeeper.properties &

或者

bin/zookeeper-server-start.sh config/zookeeper.properties 1> /dev/null &

启动kafka服务器：

bin/kafka-server-start.sh config/server.properties &

如果启动kafka服务时出现错误：

解决办法：

输入查看本机名称命令：hostname

编辑文件/etc/hosts:vi /etc/hosts

在最后一行加入：127.0.0.1  hostname命令得到的具体值

重启Kafka

我们用一个分区创建一个名为“test”的主题，只有一个副本：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

如果我们运行bin/kafka-topics.sh -list --zookeeper localhost:2181 命令，我们现在可以看到该主题：

除了手动创建topic，还可以配置broker(Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker)让它自动创建topic.

Kafka附带一个命令行客户端，它将从文件或标准输入中输入，并将其作为消息发送到Kafka群集。默认情况下，每行将作为单独的消息发送。

运行生产者，然后在控制台中输入一些消息以发送到服务器。

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

hello word

this is test send message

Kafka还有一个命令行消费者，将把消息转储到标准输出。

kafka_2.10-0.10.2.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

此时可以看到：

如果您将上述每个命令都运行在不同的终端中，那么您现在应该可以在生产者终端中输入消息，并看到它们出现在消费者终端中。
所有命令行工具都有其他选项; 运行没有参数的命令将显示更详细地记录它们的使用信息。

到目前为止，我们一直在针对一个broker，但这没有乐趣。对于Kafka，单个代理只是一个大小为1的集群，所以没有什么改变，除了启动更多的代理实例。但是为了让它感觉到，我们将集群扩展到三个节点（仍然在本地机器上）。

首先我们为每个broker设置一个配置文件（在Windows上使用copy命令）：

cp config/server.properties config/server-1.properties

cp config/server.properties config/server-2.properties

现在编辑这些新文件并设置以下属性：

vi config/server-1.properties

查找broker.id: /broker.id 回车

输入i编辑：将broker.id改为1

Esc

查找listeners: /listeners 回车

下面加入:listeners=PALINTEXT://9093

Esc

查找log.dirs: /logs.dir

修改为log.dirs=/tmp/kafka-log-1

Esc

保存修改：ZZ

vi config/server-2.properties
查找broker.id: /broker.id 回车
输入i编辑：将broker.id改为2
Esc
查找listeners: /listeners 回车
下面加入:listeners=PALINTEXT://9094

Esc

ZZ

该broker.id属性是集群中每个节点的唯一和永久名称。我们必须覆盖端口和日志目录，只因为我们在同一台机器上运行这些目录，我们希望保持broker不要在同一个端口上注册或覆盖对方的数据。

我们已经有Zookeeper，我们的单个节点启动了，所以我们只需要启动两个新节点：

bin/kafka-server-start.sh config/server-1.properties > /dev/null &

bin/kafka-server-start.sh config/server-2.properties > /dev/null &

输入jobs命令：

可以看到有一个zookeeper 、 三个broker服务在后台运行

现在创建一个复制因子为三的新主题：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic test3

执行bin/kafka-topics.sh --lsit zookeeper localhost:2181查看目前所有broker

好的，现在我们有一个集群，我们怎么知道哪个broker在做什么？运行命令：

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test3

以下是输出的说明。第一行给出了所有分区的摘要，每个附加行提供有关一个分区的信息。由于我们这个主题只有一个分区，只有一行。

“Leader”: 是负责给定分区的所有读取和写入的节点。每个节点将成为随机选择的分区部分的引导者。
“Replicas”: 是复制此分区的日志的节点列表，无论它们是leader还是目前都是活着的。
“Isr”: 是一组“同步”副本。这是副本列表的子集，该列表目前是活跃的，并且被追加到leader。

请注意，在我的示例中，节点1是主题唯一分区的领导者。

我们可以在我们创建的原始主题上运行相同的命令来查看它的位置：

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

所以没有什么惊喜 - 原来的主题没有复制品，在服务器0，我们集群中唯一的服务器。

让我们发布一些消息给我们的新主题：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test3

现在我们来看看消息：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic test3

现在让我们来测试容错。broker1作为领导者，所以我们来杀了它：

jobs

kill %3

leader已经切换到其中一个从站，节点1不再处于同步副本集中：

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test3

Leader变为0,我们再把0的那个服务杀死看看

目前只剩下zookeeper 和server-2.properties两个后台服务

果然，Leader变为2，达到容错的目的

测试发送消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --from-beginning --topic test3

测试接收消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --from-beginning --topic test3

从控制台编写数据并将其写回到控制台是一个方便的开始，但您可能希望使用其他来源的数据或将数据从kafka导出到其他系统。对于许多系统，不用编写自定义集成代码，您可以使用Kafka Connect导入或导出数据。

Kafka Connect是Kafka包含的工具，用于将数据导入和导出到Kafka。它是一种运行连接器的可扩展工具 ，它实现与外部系统交互的自定义​​逻辑。在这个快速启动中，我们将看到如何使用从文件导入数据到Kafka主题(topics)并将数据从Kafka主题导出到文件的简单连接器运行Kafka Connect。

首先，我们将首先创建一些种子数据来测试：

echo -e "foo\nbar" > test.txt

接下来，我们将启动以独立模式运行的两个连接器，这意味着它们在单个本地专用process中运行。我们提供三个配置文件作为参数。第一个是Kafka Connect进程的配置，包含常见配置，如连接的Kafka broker和数据的序列化格式。剩下的配置文件都指定要创建的连接器。这些文件包括唯一的连接器名称，要实例化的连接器类以及连接器所需的任何其他配置。

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。我们可以通过检查输出文件的内容来验证数据是否通过整个流水线传送：

cat test.sink.txt 

请注意，数据存储在Kafka主题中connect-test，因此我们还可以运行控制台消费者来查看主题中的数据（或使用自定义消费者代码来处理它）：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-begin

连接器继续处理数据，因此我们可以将数据添加到文件中，并通过管道移动：

echo "next row" >> test.txt

您应该看到行显示在控制台消费者输出和接收器文件中。

Kafka Streams是Kafka的客户端库，用于实时流处理和分析存储在kafka broker中的数据。此快速入门示例将演示如何运行在此库中编码的流应用程序。这是WordCountDemo示例代码的要点（转换为使用Java 8 lambda表达式以便于阅读）。

// Serializers/deserializers (serde) for String and Long types
final Serde<String> stringSerde = Serdes.String();
final Serde<Long> longSerde = Serdes.Long();

// Construct a `KStream` from the input topic ""streams-file-input", where message values
// represent lines of text (for the sake of this example, we ignore whatever may be stored
// in the message keys).
KStream<String, String> textLines = builder.stream(stringSerde, stringSerde, "streams-file-input");

KTable<String, Long> wordCounts = textLines
    // Split each text line, by whitespace, into words.
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))

// Group the text words as message keys
    .groupBy((key, value) -> value)

// Count the occurrences of each word (message key).
    .count("Counts")

// Store the running counts as a changelog stream to the output topic.
wordCounts.to(stringSerde, longSerde, "streams-wordcount-output");

它实现了WordCount算法，该算法从输入文本中计算单词出现直方图。但是，与以前在有限数据上操作的其他WordCount示例不同，WordCount演示应用程序的行为略有不同，因为它被设计为在无限的无界数据流上运行。与有界变体类似，它是一种有状态的算法，可以跟踪和更新单词的计数。然而，由于它必须承担潜在的无限制的输入数据，所以它会周期性地输出其当前状态和结果，同时继续处理更多的数据，因为它不知道何时处理了“全部”输入数据。

作为第一步，我们将为Kafka主题准备输入数据，随后由Kafka Streams应用程序处理。

echo -e "all streams lead to kafka\nhello kafka steams\njoin kafka summit" > file-input.txt

接下来，我们使用控制台制作者将此输入数据发送到名为streams-file-input的输入主题，该控制台生产者逐行读取STDIN中的数据，并将每行作为单独的Kafka消息以空值和值编码字符串的主题（实际上，流数据可能会连续流入Kafka应用程序将启动和运行）：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic streams-file-input

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-file-input < file-input.txt

我们现在可以运行WordCount演示应用程序来处理输入数据：

bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo

演示应用程序将从输入主题streams-file-input读取，对每个读取的消息执行WordCount算法的计算，并将其当前结果持续写入输出主题streams-wordcount-output。因此，除了记录条目之外，不会有任何STDOUT输出，因为结果被写回卡夫卡。该演示将运行几秒钟，然后，与典型的流处理应用程序不同，它会自动终止。

我们现在可以通过从其输出主题中读取来检查WordCount演示应用程序的输出：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
 --topic streams-wordcount-output \
 --from-begin \
 --formatter kafka.tools.DefaultMessageFormatter \
 --property print.key=true \
 --property print.value=true \
 --property key,deserializer=org.apache.kafka.serialization.StringDeserializer \
 --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer

将以下输出数据打印到控制台：

这里，第一列是java.lang.String格式的Kafka消息键，第二列是java.lang.Long格式的消息值。请注意，输出实际上是一个连续的更新流，其中每个数据记录（即上面的原始输出中的每一行）是单个字的更新计数，也称为“kafka”的记录键。对于具有相同键的多个记录，每个后续记录是上一个记录的更新。下面的两幅图表说明了幕后实际发生的情况。第一列显示了当前状态的KTable<String, Long>演进，即计算出现的字数count。第二列显示从状态更新到KTable而导致的更改记录，并将其发送到输出Kafka主题streams-wordcount-output。首先，正在处理文本行“所有流导致卡夫卡”。将KTable被建成为一个新的表项（以绿色背景高亮显示），每一个新词结果，并有相应的变化记录发送到下游KStream。当处理第二条文本行“hello kafka streams”时，我们首次观察到KTable正在更新的现有条目（此处为“kafka”和“streams”）。此外，更改记录正在发送到输出主题。等等（我们跳过第三行如何处理的图示）。这就解释了为什么输出主题具有我们上面显示的内容，因为它包含完整的更改记录。超越这个具体例子的范围，Kafka Streams在这里做的是利用表和changelog流之间的二元性（这里：table = KTable，changelog stream =下游的KStream）：您可以发布表到流，如果您从头到尾消耗整个changelog流，则可以重建表的内容。现在，您可以将更多的输入消息写入流文件输入主题，并观察添加到流的更多消息- 字数输出主题，反映更新的字数（例如，如上所述使用控制台生产者和控制台消费者）。您可以通过Ctrl-C停止控制台消费者。