一、kafka安装（kafka_2.9.2-0.8.1.1.zip）

1、下载并解压安装包

tar -xvf kafka_2.9.2-0.8.1.1.tgz 或者 unzip kafka_2.9.2-0.8.1.1.zip

2、修改配置文件conf/server.properties：

broker.id=0；
host.name=xxx.xxx.xxx.xxx
zookeeper.connect= xxx.xxx.xxx.xxx可逗号分隔配置多个

3、修改配置文件vim log4j.properties（最新版则不需要修改）

① log4j.appender.D.File = /data1/home/shaka/kafka/kafka_2.9.2-0.8.1.1/logs/debug.log
② log4j.appender.E.File = /data1/home/shaka/kafka/kafka_2.9.2-0.8.1.1/logs/error.log

4、启动kafka服务

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

5、查看是否启动： jsp -l

6、创建topic主题：

bin/kafka-topics.sh --create --zookeeper 主机名:2181 --replication-factor 1 --partitions 2 --topic mytopic

7、启动consumer
bin/kafka-console-consumer.sh --zookeeper 主机名:2181 --topic mytopic
（进入等待状态）

8、开启另外一个终端连接主机，启动productor

bin/kafka-console-producer.sh --broker-list <主机名>:9092 --topic mytopic
（进入等待状态）

9、开启第三个终端连接主机，发送脚本shelllogger.sh
编辑脚本，注释#log=”….”在下面添加
log=”INSERT INTO test(user,content) VALUES(‘xxxx’，’xxxxx’);”保存

10、运行脚本
./shelllogger.sh &

11、查看脚本是否运行：
tail -f access.log
（动态显示日志内容）

12、将日志实时传送到kafka指定主题中

nohup tail -f /home/storm/app/kafka_2.10-0.9.0.1/access.log | bin/kafka-console-producer.sh --broker-list mystorm:9092 --topic topic-003-name >logs/producer.log 2>&1 &
（此时ConsoleConsumer端会接收到日志信息说明成功！）

附：停止持续输出日志信息操作

jps
kill -USR2 [ConsoleProducer的进程号PID]
ps -l
kill -USR2 [shelllogger.sh进程的PID]