kafka安装

1.下载安装

wget http://mirrors.cnnic.cn/apache/kafka/0.8.1.1/kafka_2.9.2-0.8.1.1.tgz

解压tar -xzvf kafka_2.9.2-0.8.1.1.tgz 之后,查看路径下的文件

[root@master intalldir]# cd kafka_2.9.2-0.8.1.1/

[root@master kafka_2.9.2-0.8.1.1]# ll

total 28

drwxr-xr-x. 3 root root 4096 Apr 22 2014 bin

drwxr-xr-x. 2 root root 4096 Apr 22 2014 config

drwxr-xr-x. 2 root root 4096 Apr 22 2014 libs

-rw-rw-r--. 1 root root 11358 Apr 22 2014 LICENSE

-rw-rw-r--. 1 root root 162 Apr 22 2014 NOTICE

配置环境变量

vim /etc/profile

然后 运行 source /etc/profile使之生效

2.配置kafka

配置以及启动测试

(1)首先可以启动kafka自带的ZooKeeper

bin/zookeeper-server-start.sh config/zookeeper.properties &（注意要后台启动）

但是最好使用自己安装的ZooKeeper，修改配置文件config/server.properties，修改如下内容：

查看server.properties

# Zookeeper connection string (see zookeeper docs for details).

# This is a comma separated host:port pairs, each corresponding to a zk

# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".

# You can also append an optional chroot string to the urls to specify the

# root directory for all kafka znodes.

zookeeper.connect=localhost:2181

利用逗号zookeeper集群

配置如下

zookeeper.connect=master:2181,slave1:2181,slave2:2181

(2)将kafka分别复制到slave1和slave2中

并修改配置文件config/server.properties内容如下所示：

broker.id=1 # 在slave1修改broker.id=2 # 在是slave2修改

因为Kafka集群需要保证各个Broker的id在整个集群中必须唯一，需要调整这个配置项的值（如果在单机上，可以通过建立多个Broker进程来模拟分布式的Kafka集群，也需要Broker的id唯一，还需要修改一些配置目录的信息）。

(3)在集群中的master、slave1、slave2这三个节点上分别启动Kafka，分别执行如下命令：（启动前保证zooKeeper已经启动）

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

(4)可以通过查看日志，或者检查进程状态，保证Kafka集群启动成功。 我们创建一个名称为my-replicated-topic5的Topic，5个分区，并且复制因子为3，执行如下命令：

bin/kafka-topics.sh --create --zookeeper h1:2181,h2:2181,h3:2181 --replication-factor 3 --partitions 5 --topic my-replicated-topic5

查看创建的Topic，执行如下命令：

bin/kafka-topics.sh --describe --zookeeper h1:2181,h2:2181,h3:2181 --topic my-replicated-topic5

结果信息如下所示：



上面Leader、Replicas、Isr的含义如下：

Partition： 分区Leader ： 负责读写指定分区的节点Replicas ： 复制该分区log的节点列表Isr ： "in-sync" replicas，当前活跃的副本列表（是一个子集），并且可能成为Leader

(5)测试生产和消费

我们可以通过Kafka自带的bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh脚本，来验证演示如果发布消息、消费消息。 在一个终端，启动Producer，并向我们上面创建的名称为my-replicated-topic5的Topic中生产消息，执行如下脚本：

bin/kafka-console-producer.sh --broker-list h1:9092,h2:9092,h3:9092 --topic my-replicated-topic5

在另一个终端，启动Consumer，并订阅我们上面创建的名称为my-replicated-topic5的Topic中生产的消息，执行如下脚本：

bin/kafka-console-consumer.sh --zookeeper h1:2181,h2:2181,h3:2181 --from-beginning --topic my-replicated-topic5

可以在Producer终端上输入字符串消息行，然后回车，就可以在Consumer终端上看到消费者消费的消息内容。 也可以参考Kafka的Producer和Consumer的Java API，通过API编码的方式来实现消息生产和消费的处理逻辑。