kafka安装与使用

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

kafka目录结构

目录	说明
bin	操作kafka的可执行脚本，还包含windows下脚本
config	配置文件所在目录
libs	依赖库目录
logs	日志数据目录，目录kafka把server端日志分为5种类型， 分为:server,request,state，log-cleaner，controller

1. 解压

> tar -xzf kafka_2.9.1-0.8.2.2.tgz
> cd kafkakafka_2.9.1-0.8.2.2

2. 配置

kafka最为重要三个配置依次为：broker.id、log.dir、zookeeper.connect

kafka server端config/server.properties参数说明参照: kafka主要配置

根据属性说明完成配置：

broker.id = 1

port = 9092

host.name=h1(h1为本人配置的hostname)

log.dirs=/root/software/kafka/log

zookeeper.connect=h1:2181

配置zookeeper(假设您已经安装了zookeeper,如果没有安装，参照 zookeeper集群环境安装配置)

3.启动服务

在启动kafka时需要先启动zookeeper服务

kafka启动命令：

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

通过jps命令可以查看zookeeper和kafka是否启动成功，如下图：



4.创建topic (此处创建是名称为‘my-topic’的topic，它只有一个分区，一个副本)

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

5.查看topic列表

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

6.发送消息

Kafka 使用一个简单的命令行producer，从文件中或者从标准输入中读取消息并发送到服务端。默认的每条命令将发送一条消息。

运行producer并在控制台中输一些消息，这些消息将被发送到服务端：

> bin/kafka-console-producer.sh --broker-list
h1:9092 --topic my-topic

this is one message

this is another message

如图：



7.启动consumer

Kafka也有一个命令行consumer可以读取消息并输出到标准输出：

> bin/kafka-console-consumer.sh --zookeeper
h1:2181 --topic my-topic --from-beginning

this is one message

this is another message

如图：

刚才只是启动了单个broker，现在启动有3个broker组成的集群，这些broker节点也都是在本机上的：
1. 首先为每个节点编写配置文件：

> cp config/server.properties config/server1.properties
> cp config/server.properties config/server2.properties

编辑新文件并设置如下配置

config/server1.properties:
broker.id=1
port=9093
log.dirs=log.dirs=/root/software/kafka/log1
config/server2.properties:
broker.id=2
port=9094
log.dirs=/root/software/kafka/log2

broker.id在集群中唯一的标注一个节点，因为在同一个机器上，所以必须制定不同的端口和日志文件，避免数据被覆盖。

2. 之前我们已经启动zookeeper和一个节点，现在只需要启动这两个新节点

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

3. 创建一个拥有三个副本的topic： replicated-topic

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

我们可以通过运行"describe topics"命令来查看每个节点信息

> bin/kafka-topics.sh --describe --zookeeper h1:2181 --topic replicated-topic

Topic:replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:
Topic: replicated-topic    Partition: 0    Leader: 0    Replicas: 0,1,2    Isr: 0,1,2

下面解释一下这些输出。第一行是对所有分区的一个描述，然后每个分区都会对应一行，因为我们只有一个分区所以下面就只加了一行。

leader：负责处理消息的读和写，leader是从所有节点中随机选择的.
replicas：列出了所有的副本节点，不管节点是否在服务中.

isr：是正在服务中的节点.
在我们的例子中，节点0是作为leader运行。

我们可以对比一下看一下之前创建的my-topic



4. 发送接收消息如上步骤6、7，不再重复写出

5. 现在测试一下多broker集群的容错能力，通过"describe topics"命令知道
replicated-topic的leader为broker0，现在我们kill掉它

> ps -ef | grep server.properties

> kill –9 20549

另外一个节点被选做了leader,node 0不再出现在 in-sync
副本列表中：



通过测试验证虽然最初负责续写消息的leader
down掉了，但之前的消息还是可以消费的：