1、zookeeper搭建
Kafka集群依赖zookeeper，需要提前搭建好zookeeper
单机模式（7步）(集群模式进阶请移步：http://blog.51cto.com/nileader/795230)
Step1：
cd /usr/local/software
wget http://download.oracle.com/otn-pub/java/jdk/8u161-b12/2f38c3b165be4555a1fa6e98c45e0808/jdk-8u161-linux-x64.tar.gz
配置JAVA环境。检验方法：执行java –version和javac –version命令。
Step2：下载并解压zookeeper。
链接：http://mirror.bjtu.edu.cn/apache/zookeeper/zookeeper-3.4.3/，（更多版本：http://dwz.cn/37HGI
）最终生成目录类似结构：/home/admin/taokeeper/zookeeper-3.4.3/bin
Step3：重命名 zoo_sample.cfg文件
mv /home/admin/taokeeper/zookeeper-3.4.3/conf/zoo_sample.cfg zoo.cfg
Step4：vi zoo.cfg，修改
dataDir=/home/admin/taokeeper/zookeeper-3.4.3/data
Step5：创建数据目录：mkdir /home/admin/taokeeper/zookeeper-3.4.3/data
-
mkdir /home/admin/taokeeper/zookeeper-3.4.3/data
Step6：启动zookeeper：执行
/home/admin/taokeeper/zookeeper-3.4.3/bin/zkServer.sh start
Step7：检测是否成功启动：执行
/home/admin/taokeeper/zookeeper-3.4.3/bin/zkCli.sh 或 echo stat|nc localhost 2181
2、下载Kafka
# mkdir -p /usr/local/software
# cd /usr/local/software
# wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka-1.0.0-src.tgz
# wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka_2.12-1.0.0.tgz
kafka_2.12-1.0.0.tgz //其中2.12-1.0.0为Scala的版本，kafka-1.0.0-src.tgz为kafka版本
3、解压
# tar zxf kafka_2.12-1.0.0.tgz -C /usr/local/
# cd /usr/local/
# mv kafka_2.12-1.0.0/ kafka/
4、配置
# vi /usr/local/kafka/config/server.properties
# broker的ID，集群中每个broker ID不可相同
broker.id=0
# 监听器，端口号和port一致即可
listeners=PLAINTEXT://:9092
# Broker的监听端口
port=9092
# 必须填写当前服务器IP地址
host.name=10.10.6.225
# 必须填写当前服务器IP地址
advertised.host.name=10.10.6.225
# 暂未配置集群
#zookeeper.connect=zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181
5、配置Kafka的环境变量
# vi /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
# source /etc/profile
6、启动Kafka
# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
官方推荐启动方式：
# /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
但这种方式退出shell后会自动断开
7、验证
# jps
2256 Jps
2237 Kafka
看到Kafka的进程，说明Kafka已经启动
8、创建topic
创建名为test，partitions为3，replication为3的topic
# kafka-topics.sh --create --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --partitions 3 --replication-factor 3 --topic test
查看topic状态
# kafka-topics.sh --describe --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test
Topic:test PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test Partition: 0 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
Topic: test Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: test Partition: 2 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1
状态说明：test有三个分区分别为1、2、3，分区0的leader是3（broker.id），分区0有三个副本，并且状态都为lsr（ln-sync，表示可以参加选举成为leader）。
删除topic
在config/server.properties中加入delete.topic.enable=true并重启服务，在执行如下命令
# kafka-topics.sh --delete --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test
9、测试使用Kafka
发送消息
# kafka-console-producer.sh --broker-list zookeeper-01:9092,zookeeper-02:9092,zookeeper-03:9092 --topic test
输入以下信息：
This is a message
This is another message
接收消息
# kafka-console-consumer.sh --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test --from-beginning //--from-beginning 表示从开始第一个消息开始接收
若看到上输入的信息说明已经搭建成功。
