kafka0.8.2集群的环境搭建并实现基本的生产消费

Kakfa起初是由LinkedIn公司开发的一个分布式的消息系统，后成为Apache的一部分，它使用Scala编写，以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如Cloudera、Apache Storm、Spark等都支持与Kafka集成。

 

此处采用kafka_2.10-0.8.2.2版本进行安装部署，机器环境为ubuntu1404

1、下载kafka安装包并解压

此处解压到/usr/local目录下

> tar -zxvf kafka_2.10-0.8.2.2.tgz -C /usr/local/
> cd /usr/local/kafka_2.10-0.8.2.2/

2、开启kafka服务

Kafka依赖zookeeper，所以首先要开启zookeeper，如果机器没有安装zookeeper，可以使用kafka安装包内的自带的脚本来启动一个单节点的zookeeper

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

启动kafka server：

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

3、创建topic

创建一个名为”test”的topic，只有一个partition和一个replica

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

使用topic命令可以查看当前的topic

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

相对于此处的手动创建topic，你还可以在brokers上配置当没有topic的时候自动创建topic

4、发送消息

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

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

5、开启consumer

Kafka还有一个命令行consumer，可以把消息转储到标准输出。

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

6、设置multi-broker集群

对kafka来说，单个broker只是一个大小为1的集群，现在把集群增加到3个（仍然在单台机器上实现）

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

编辑这3个配置文件

config/server-1.properties:
broker.id=1
port=9093
log.dir=/tmp/kafka-logs-1

config/server-2.properties:
broker.id=2
port=9094
log.dir=/tmp/kafka-logs-2

config/server-3.properties:
broker.id=3
port=9095
log.dir=/tmp/kafka-logs-3

broker.id
是集群中每个节点的唯一和永久名称，zookeeper已经启动，我们需要开启3个节点

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

创建一个replication为3 的新topic

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

第一行给出所有partitions的概览，每个附加行给出一个分区的信息，由于我们只有一个分区，所以此处只有一行

“leader”是负责给定分区的所有读写操作的节点，每个节点将成为随机选择的分区部分的引导者。

“replicas” 是复制此分区的日志的节点列表，无论这些节点是否是leader或者是否还存在。

"isr"是一组in-sync"副本，这是副本列表的子集，该列表目前正在生效并被领导者所追踪

此处，node 3是该topic唯一partition的leader

对之前的”test” 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

显然，”test” topic没有replicas 

发送几条数据到新的topic

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

消费这几条信息：

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

接下来测试一下容错，Broker 3 作为leader，我们将其kill掉

> ps -ef | grep server-3.properties
root       8322   6862  1 16:21 pts/19   00:00:35 /usr/lib/jvm/java-8-oracle/bin/java -Xmx1G -Xms1G –server………………….
> kill -9 8322

Leader切换到了之前的某一个slave，node 3 不再处于同步复制集

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

但是消息依然可以被消费，即使之前负责写的leader已经挂了

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

参考：http://kafka.apache.org/082/documentation.html#quickstart