- 目录
- 1、Kafka集群部署
- 2、Kafka常用操作命令
目录
1、Kafka集群部署
1.1、下载安装包
http://kafka.apache.org/downloads.html
在linux中使用wget命令下载安装包
wget http://mirrors.hust.edu.cn/apache/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz
1.2、解压安装包
tar -zxvf /export/software/kafka_2.11-0.8.2.2.tgz -C /export/servers/
1.3、修改配置文件
vi /export/servers/kafka/config/server.properties
输入以下内容：
#broker的全局唯一编号，不能重复
broker.id=0#用来监听链接的端口，producer或consumer将在此端口建立连接
port=9092#处理网络请求的线程数量
num.network.threads=3#用来处理磁盘IO的线程数量
num.io.threads=8#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400#接受套接字的缓冲区大小
socket.receive.buffer.bytes=102400#请求套接字的缓冲区大小
socket.request.max.bytes=104857600#kafka运行日志存放的路径
log.dirs=/root/kafkalog#topic在当前broker上的分片个数
num.partitions=2#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1#segment文件保留的最长时间，超时将被删除
log.retention.hours=168#滚动生成新的segment文件的最大时间
log.roll.hours=168#日志文件中每个segment的大小，默认为1G
log.segment.bytes=1073741824#周期性检查文件大小的时间
log.retention.check.interval.ms=300000#日志清理是否打开
log.cleaner.enable=true#broker需要使用zookeeper保存meta数据
zookeeper.connect=shizhan:2181,mini2:2181,mini3:2181#zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000#partion buffer中，消息的条数达到阈值，将触发flush到磁盘
log.flush.interval.messages=10000#消息buffer的时间，达到阈值，将触发flush到磁盘
log.flush.interval.ms=3000#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
delete.topic.enable=true#此处的host.name为本机IP(重要),如果不改,则客户端会抛出:Producer connection to localhost:9092 unsuccessful 错误!
host.name=192.168.112.2001.4、分发安装包
scp -r /export/servers/kafka_2.11-0.8.2.2 kafka02:/export/servers
1.5、再次修改配置文件（重要）
依次修改各服务器上配置文件的的broker.id，分别是0,1,2不得重复。
对应的host的IP地址更改为各个主机的ip地址
另外将产生的log文件的输出地址更改下
zk的地址更改为自己机器的地址
1.6、启动集群
依次在各节点上启动kafka
bin/kafka-server-start.sh config/server.properties
2、Kafka常用操作命令
- 查看当前服务器中的所有topic
bin/kafka-topics.sh –list –zookeeper zk01:2181 - 创建topic
./kafka-topics.sh –create –zookeeper mini1:2181 –replication-factor 1 –partitions 3 –topic first - 删除topic
sh bin/kafka-topics.sh –delete –zookeeper zk01:2181 –topic test
需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。 - 通过shell命令发送消息
kafka-console-producer.sh –broker-list kafka01:9092 –topic itheima - 通过shell消费消息
sh bin/kafka-console-consumer.sh –zookeeper zk01:2181 –from-beginning –topic test1 - 查看消费位置
sh kafka-run-class.sh kafka.tools.ConsumerOffsetChecker –zookeeper zk01:2181 –group testGroup - 查看某个Topic的详情
sh kafka-topics.sh –topic test –describe –zookeeper zk01:2181
详解)
