1.默认安装好zookeeper和scala
2.下载安装包,解压
tar -zxvf kafka_2.11-0.9.0.1.tgz kafka_2.11-0.9.0.1
3.配置环境变量
vim /etc/profile
#kafka
export KAFKA_HOME=/opt/kafka_2.11-0.9.0.1
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
4.修改配置文件
config/server.properties
broker.id=0 //整型 从0开始 高可用是注意修改
host.name=127.0.0.1 //节点IP 高可用时注意修改本机IP 172.16.104.241
log.dirs=/opt/kafka_2.11-0.9.0.1/logs //日志目录
zookeeper.connect=127.0.0.1:2181 //zookeeper 多个时用逗号分隔 高可用时为zookeeper集群 A:9092,B:9092
zookeeper.connection.timeout.ms=600000 //连接超时
config/producer.properties
metadata.broker.list=127.0.0.1:9092 //注意修改IP地址 高可用时为本机Ip A:9092,B:9092
config/consumer.properties
zookeeper.connect=127.0.0.1:2181 //注意修改IP地址 高可用时为zookeeper集群 A:9092,B:9092
5.验证
1.启动zookeeper
zkServer.sh start //高可用时 需每台机子执行一次
2.启动kafka server端
bin/kafka-server-start.sh config/server.properties & //高可用时 需每台机子执行一次
3.启动producer
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test01
4.启动consumer
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test01 --from-beginning
//注意集群环境时需更改为zookeeper 集群链接 如: A:2181,B:2181,C:2181.....
5.此时在producer中输入内容在consumer中可以显示该内容(需多开几个命令行窗口)
6.示例使用(//注意集群环境时需更改为zookeeper 集群链接 如: A:2181,B:2181,C:2181.....)
1.创建主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
// 高可用时 replication-factor为副本数 不能多于broker节点数 partitions 为分区数
2.查看主题
bin/kafka-topics.sh --list --zookeeper localhost:2181
3.查看主题详情
bin/kafka-topics.sh--describe --zookeeper localhost:2181 --topic test
4.删除主题
bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper localhost:2181 --topic test
5.同步文件内容
bin/connect-standalone.sh config/connect-standalone.properties
config/connect-file-source.properties config/connect-file-sink.properties
//将connect-file-source.properties配置的文件内容同步到connect-file-sink.properties配置的文件中