1,下载并解压kakfa

地址 http://kafka.apache.org/downloads.html
解压 kafka_2.11-2.1.0.tgz

2,kafka集群需要zookeeper

下载zookeeper并解 http://apache.fayea.com/zookeeper/

3,编辑kafka,配置文件

进入解压后的kafka目录，修改server.propertis文件
设置listeners=PLAINTEXT://192.168.26.130:9092（当前机器ip）
其它参数默认值

4, 启动kafka

先启动zookeeper,进入zookeeper的bin目录，执行./zkService.sh start
启动kafka 进入kafka bin目录，执行./kafka-server-start.sh ../config/server.properties 
。

5,查看 zookeeper和kafka是否启动成功

创建一个topic : 
./kafka-topics.sh --zookeeper localhost:2181 --create --topic test4 --partitions 1 --replication-factor 1 
表示 test4 1分区，1个副本，
查看topic list : 
./kafka-topics.sh --zookeeper localhost:2181 --list ;可查看到上面 刚创建的 test4
启动producer 打开新的终端 : 
bin/kafka-console-producer.sh --broker-list 192.168.1.104:9092 --topic test4
输进消息： lmx
启动consumer 打开新的终端： 
./kafka-console-consumer.sh --bootstrap-server 192.168.26.130:9092 --topic test4 --from-beginning
测试： 
producer终端输入 hello kafka，查看consumer终端是否接口到消息

6、kafka相关命令

1.console窗口:

producer:

./kafka-console-producer.sh --broker-list  localhost:9092  --topic my_test_topic

consumer:

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my_test_topic --from-beginning

2.查看消费情况:

列出所有consumer group:

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

查看组消费情况:

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my_group_02
 

./kafka-consumer-groups.sh --zookeeper localhost:2181 --describe --group my_group_02

查看组中存活的consumer

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my_group_02 --members

查看组中存活的consumer消费的partition

 ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my_group_02 --members --verbose

3.查看log文件中消息内容:

server.properties 配置文件中log.dirs配置项即为log文件所在路径

到bin目录下执行:

./kafka-run-class.sh kafka.tools.DumpLogSegments --files 目标log文件 --print-data-log