大数据之Kafka集群安装及简单使用

水深无声 2022-05-28 14:06 189阅读 0赞

*  目录
    
     *  1、Kafka集群部署
     *  2、Kafka常用操作命令

# 目录 #

## 1、Kafka集群部署 ##

**1.1、下载安装包**  
[http://kafka.apache.org/downloads.html][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][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.200

**1.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

[http_kafka.apache.org_downloads.html]: http://kafka.apache.org/downloads.html
[http_mirrors.hust.edu.cn_apache_kafka_0.8.2.2_kafka_2.11-0.8.2.2.tgz]: http://mirrors.hust.edu.cn/apache/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz