Kafka集群的配置与部署

本假设你本机还没有安装任何的Zookeeper节点或者Kafka节点，首先我们需要下载Kafka源代码然后将其解压缩:

> tar -xzf kafka_2.11-0.9.0.0.tgz
> cd kafka_2.11-0.9.0.0

Kafka默认使用ZooKeeper作为服务协调器，并且Kafka内置了一个ZooKeeper的运行脚本，可以让你快速地启动允许一个单节点的ZooKeeper实例:

> bin/zookeeper-server-start.sh config/zookeeper.properties
[2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
...

然后启动Kafka服务器:

> bin/kafka-server-start.sh config/server.properties
[2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)
[2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)
...

使用

在启动了Kafka服务器之后，我们首先来尝试创建一个名为test的主题，设置只有一个Partition与一个Replica。

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

然后我们在查询主题列表的时候就可以看到该主题了:

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

除了手动地指定创建主题，我们也可以配置Brokers在收到不存在的主题的时候自动创建该主题。在Topics创建好之后，就可以通过其来发送消息。Kafka为我们提供了一个命令行工具，可以允许我们从某个文件或者其他标准的输入流中抓取数据然后将其以消息形式发送到Kafka集群，默认情况下每个单独的行会被认为是一条单独的消息。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
This is a message
This is another message

消息发送完毕之后，我们需要设置Consumer端来读取消息，Kafka同样为我们提供了这样一个便捷的命令行工具，

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
This is a message
This is another message

如果你在不同的命令行中分别运行上述命令，那么你的Consumer已经能够接收到这些数据了。