Kafka 入门指南

本文是 Kafka 的入门指南。

kafka(http://kafka.apache.org/) 非常强大，应用场景也很多，性能很好。

基本要求

1. JDK
2. zookeeper

单机单点安装测试

第一步：下载解压

下载地址 http://kafka.apache.org/downloads.html

tar -xzf kafka_2.11-0.11.0.1.tgz cd kafka_2.11-0.11.0.1

第二步：启动 zookeeper-server 和 kafka-server

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

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

关于 server.properties 配置文件请参考后文：《server.properties配置文件参数说明》

第三步：新建一个 topic 创建一个名为 “test” 的 Topic，只有一个分区和一个备份：

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

创建好之后，可以通过运行以下命令，查看已创建的 topic 信息：

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

第四步：发送消息

Kafka 提供了一个命令行的工具，可以从输入文件或者命令行中读取消息并发送给Kafka集群。每一行是一条消息。运行 producer（生产者）,然后在控制台输入几条消息到服务器。

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

1
2

>This is a message
>This is another message

第五步：消费消息

Kafka 也提供了一个消费消息的命令行工具，将存储的信息输出出来。

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –topic test –from-beginning 或者 bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –from-beginning

This is a message This is another message

如果你有2台不同的终端上运行上述命令，那么当你在运行生产者时，消费者就能消费到生产者发送的消息。 所有的命令行工具有很多的选项，你可以查看文档来了解更多的功能。

单机多点安装

第六步：设置有多个broker的集群（单机）

到目前，我们只是单一的运行一个 broker ，没什么意思。对于 Kafka ,一个 broker 仅仅只是一个集群的大小, 所有让我们多设几个 broker 。注意：broker.id 从 0 开始递增，每个 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 是集群中每个节点的唯一永久的名称，我们修改端口和日志分区是因为我们现在在同一台机器上运行，我们要防止 broker 改写同一端口上注册的数据。

我们已经运行了 zookeeper 和刚才的一个 kafka 节点(备注： broker.id=0 也要运行启动。)，所有我们只需要在启动2个新的kafka节点。

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

现在，我们创建一个新 topic ，把备份设置为：3

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

好了，现在我们已经有了一个集群了，我们怎么知道每个 broker 在做什么呢？ 运行命令 describe topics

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: 0 Replicas: 0,1,2 Isr: 1,2,0

这是一个解释输出，第一行是所有分区的摘要，下面的每行提供一个分区信息，因为我们只有一个分区，所有只有一行。

“leader”：该节点负责所有指定的分区读和写，每个节点都可能领导一个随机选择的分区。 “replicas”:备份的节点，无论该节点是否活着,只是显示。 “isr”：备份节点的集合，也就是活着的节点集合。

我们运行这个命令，看看一开始我们创建的那个节点。

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

没有惊喜，原始主题没有 Replicas，所以是 0 。

让我们来发布一些信息在新的 topic 上：

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

现在，消费这些消息。

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –from-beginning –topic my-replicated-topic … my test message 1 my test message 2

我们要测试集群的容错，kill 掉 leader，Leader: 0 表示当前的 leader 是 Broker.id=0 ，也就是 kill 掉 Broker0。

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

备份节点之一成为新的 leader ，而 broker0 已经不在同步备份集合里了。

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

但是，消息没仍然没丢：

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –from-beginning –topic my-replicated-topic … my test message 1 my test message 2 my test message 3 my test message 4 …

Kafka Consumer Group

如何在Kafka的命令行创建多个消费者？

./bin/kafka-console-consumer.sh –zookeeper localhost:2181 –topic test –from-beginning –consumer-property group.id=local

server.properties配置文件参数说明

每个kafka broker中配置文件server.properties默认必须配置的属性如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

broker.id=0
num.network.threads=2
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=2
log.retention.hours=168

log.segment.bytes=536870912
log.retention.check.interval.ms=60000
log.cleaner.enable=false

zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=1000000

参数说明

kafka中Leader,replicas配置参数

kafka中zookeeper参数配置

参考资料

1. Kafka
2. Kafka quickstart
3. kafka quickstart 的中文翻译
4. 如何为Kafka集群选择合适的Topics/Partitions数量？
5. create-multiple-consumers-in-kafka-in-command-line
6. Kafka 指南

茶歇驿站

一个可以让你停下来看一看，在茶歇之余给你帮助的小站。

这里的内容主要是后端技术，个人管理，团队管理，以及其他个人杂想。