1.2 使用案例

消息处理

kafka是一个很好的传统消息代理（message broker）替代产品。使用消息代理有几种原因：解耦生产者与消息处理、缓存消息等。与大多数消息系统相比，kafka有更好的吞吐量，内置分区，复制和容错性，这使它成为大规模消息处理应用很好的解决方案。

根据我们的经验，消息传递的使用往往是相对较低的吞吐量，但可能需要较低的端到端延迟，并且通常取决于Kafka提供的强大的持久性保证。

在这个领域，Kafka与传统的消息系统（如ActiveMQ或RabbitMQ）相当。

网站行为跟踪
kafka的原始用例（为此而生）是能重建一套可以实时发布，实时订阅消息，用于处理用户活动轨迹跟踪的管道。也就是说网站的活动（页面浏览、搜索、用户其它行为）可以按活动类型分别发布到各自的主题；这些订阅可以被用于后续各种用途：包括实时处理、实时监控、加载到hadoop、离线数据仓库以进行离线处理和生成报告。
因为每个用户浏览页面都会产生活动消息，因此，活动跟踪数据量非常大。

Metrics度量

Kafka经常被用于处理监控数据。这涉及到从分布式应用收集统计数据，并且做为后续分析的一个统一的数据源。（即分布式统计数据查询入口或代理）

日志收集

很多人把kafka做为日志收集解决方案的替代品。日志收集是从服务器上采集日志文件并把它们放入一个集中位置（如：文件服务器或hdfs）进行处理。kafka抽象了文件细节，并将日志或事件数据作为消息流进行更清晰的抽象。这样可以实现更低延迟的处理，更容易支持多数据源以及分布式消息处理。与Scribe和Flume相比，kafka提供同样的良好性能，并提供更好的可用性（因为多个副本），和更低的延时。

流处理

Kafka的许多用户在处理管道中处理数据，这些数据由多个阶段组成，其中原始输入数据从Kafka topic中消费，然后聚合，补充或以其他方式转化为新的topic，供进一步消费或后续处理。通过kafka topics串联起所有步骤，形成一个数据处理通道。
如：一个处理新闻的流程：首先通过RSS收集新闻，并发布到”articles”主题中；第二步，从“articles”主题中取新闻并清洗重复内容，然后发布一个新的主题中；最后，从上步的主题中取数据并推荐给用户。
这样的处理管道是基于单个主题的实时数据流程图。从0.10.0.0版本开始，一个轻量但强大的，被称为kafka stream的功能用于处理这样的数据。除了Kafka stream还有另外相似的开源工具：Apache Storm / Apache Samza。

事件溯源（Event Sourcing）

事件溯源是应用程序设计的一种风格，其中状态更改以时间排序的记录序列进行记录。 Kafka对非常大的存储日志数据的支持使得它成为以这种风格构建的应用程序的优秀后端。

提交日志Commit Log

kafka可以做为分布式系统的外部提交日志服务器。可以帮助分布式节点存储数据失败时，做为重新同步机制，在节点与操作之间复制日志，以恢复数据。
这种情况，kafka与Apache BookKeeper 非常相似。

1.3 快速开始

本教程，假设你没有任何kafka知识。并且没有现成的kafka和zookeeper数据。

Step 1: 下载代码

下载0.11.0.2版本代码，并且解压

1 > tar -xzf kafka_2.11-0.11.0.2.tgz
2 > cd kafka_2.11-0.11.0.2

Step 2: 启动服务

kafka依赖zookeeper，因此首先要启动zookeeper；如果没有安装独立的zookeeper，可以使用kafka内嵌的zookepper。虽然这种方式快速但不是很好。
启动zookeeper

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

启动 kafka

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

Step 3: 创建一个主题

手动创建一个名为“test”的主题

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

查看刚创建的主题

bin/kafka-topics.sh --list --zookeeper localhost:2181

可以通过配置“自动创建主题”，这样如果没有提前创建主题，那么在发布消息时，如果此消息对应的主题不存在，会自动创建。

Step 4: 发送消息

通过命令行客户端，可以通过文件或标准输入（命令行）向kafka集群发送消息。默认每行都是一条消息。
启动生产者（启动成功进入命令行阻塞状态，可以输入数据，回车发送）

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

Step 5: 启动消费者

启动消费者命令行(启动后命令行处于阻塞状态，生产者发布的消息会在此显示)

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

Step 6: 设置服务器集群

到目前为止，我们一直在使用一个broker，但这并不好玩。 对于kafka来说，一个broker是只是有一台服务的集群，所以集群除了多启动几个broker实例之外，没有太大的改变。。 现在来感受下，让我们把我们的集群扩展到三个节点（仍然都在我们的本地机器上）。

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

现在编辑这些新文件并设置以下属性：
config/server-1.properties:

broker.id=1
listeners=PLAINTEXT://:9093
log.dir=/tmp/kafka-logs-1

config/server-2.properties:

broker.id=2
listeners=PLAINTEXT://:9094
log.dir=/tmp/kafka-logs-2

broker.id属性是群集中每个节点的唯一且永久的名称。 我们必须重写端口和日志目录，因为我们在同一台机器上运行这些端口和日志目录，我们希望让代理不要尝试在同一个端口上注册或覆盖彼此的数据。

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

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

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

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

现在我们已经创建一个集群，但是我们怎么知道每个broker都做了什么？执行如下命令：”describe toics”：

>bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0

这里是对输出的解释。 第一行给出了所有分区的摘要，每个附加行给出了关于一个分区的信息。 由于我们只有一个分区，所以只有一行。

• “leader” 是负责该节点指定分区的所有读写操作。如果一个节点被分配（随机的）了一个分区，则这个节点会是这个分区的leader。上图可以看出分区leader的broker.id是1，可以唯一一个，因为分区数是1。
• “replicas” 是分区副本保存节点的列表，无论这些备份节点是否leader，也不管这些节点是否活着，上面可以看出在节点1,2,0上各保存了一个副本。
• “isr” 是同步的副本。这是副本列表的子集，即当前还活着的并且可以被leader联系到的。

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

我们可以运行相同的命令行，用来查看我们最初创建的topic的信息

>bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test PartitionCount:1 ReplicationFactor:1 Configs:
Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0

毫无疑问，最初的topic没有副本备份，并且处于server 0上，就是当我们创建这个server时，我们的集群中仅有的server。

下面向新topic发送一些消息：

>bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
...
my test message 1
my test message 2
^C

然后消费这些消息：

>bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C

现在可以测试kafka 的容错性能，broker 1一直充当leader，我们可以杀掉他：

>ps aux | grep server-1.properties
7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java...
kill -9 7564

leader已经切换到两个从节点中的一个，并且broker1已经不在ISR中：

>bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 1,2,0 Isr: 2,0

即使当初写入的leader已经被killed，但是消息仍然可以使用

>bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C

Step7：使用 Kafka Connect导出/导入数据

通过命令行读写数据是一个方便简单的开始。但是你可能希望通过其它数据源或者从kafka导出数据到其它系统。对大多数系统，你不需要写定制化的代码，只需要使用kafka connect 就可以导入或导出数据。kafka connect是一个可以运行多个connectors的扩展工具集。这些connectors实现了与外部系统交互的逻辑。在这个示例中，我们将体验怎么运行kafka connect ,并通过简单的connectors从文件把数据导入到主题，并从主题中把数据导出到文件。

首先，我们创建一些种子数据，如下：

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

接下来，我们在独立模式下启动运行两个connectors，也就是在单一的，本地，专用的线程中。我们提供了三个配置文件作为参数。第一个是 Kafka Connect进程的，包含一些常见的配置，比如要链接的brokers和要序列格式化的数据。其余的每个配置都指定一个连接器去创建。这些文件包括一个唯一的连接器名称，连接器实例化类，和一些连接器需要的其他配置。

> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

kafka包内包含了这些配置文件，用默认本地的kafka集群配置，并且创建两个connectors：第一个是一个source connector用来从输入文件逐行读取并且发送到topic，第二个是sink connector从topic读取消息并且逐行写到文件中。

当启动过程中，在会看到连接器的实例化日志。启动成功后，source connector开始从test.txt读取数据并且发送到connect-test这个topic中。同时sink connector开始从connect-test topic中读取消息并且写到test.sink.txt文件中。
我们可以看下test.sink.txt这个文件，看有没有数据写入：

> cat test.sink.txt foo bar
foo
bar

该数据被存储在kafka的connect-test topic中，所以我们可以运行一个消费者终端去看看在这个topic中的数据：

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"} {"schema":{"type":"string","optional":false},"payload":"bar"} ...

connectors 继续处理数据，我们尝试向test.txt文件中继续添加数据：

> echo "Another line" >> test.txt

你应该看到有一行在消费者控制台输入，并且写入了sink文件。

Step 8：使用Kafka流来处理数据

见第9章