Kafka的安裝和使用

kafka的安裝

第一步：安裝zookeeper，kafka依賴zookeeper，所以需要先安裝zookeepe

第二步：下載安裝kafka
wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz
tar -xzf kafka_2.12-2.8.1.tgz
cd kafka_2.12-2.8.1
第三步：修改配置
修改配置文件 config/server.properties

#broker.id屬性在kafka集群中必須要是唯一
broker.id=0
#kafka部署的機器ip和提供服務的端口號
listeners=PLAINTEXT://192.168.65.60:9092   
#kafka的消息存儲文件
log.dir=/kafka/kafka-logs
#kafka連接zookeeper的地址
zookeeper.connect=192.168.65.60:2181

第四步：啟動服務

啟動腳本語法：kafka-server-start.sh [-daemon] server.properties
可以看到，server.properties的配置路徑是一個強制的參數，-daemon表示以后台進程運行，否則ssh客戶端退出后，就會停止服務。
(注意，在啟動kafka時會使用linux主機名關聯的ip地址，所以需要把主機名和linux的ip映射配置到本地host里，用vim /etc/hosts)

查看主機名稱
hostname 查看主機名
hostname -i：查看本機對應的IP
修改主機名稱：vim /etc/hostname

# 啟動kafka，運行日志在logs目錄的server.log文件里
bin/kafka-server-start.sh -daemon config/server.properties   #后台啟動，不會打印日志到控制台
或者用
bin/kafka-server-start.sh config/server.properties &

# 我們進入zookeeper目錄通過zookeeper客戶端查看下zookeeper的目錄樹
bin/zkCli.sh
ls /        #查看zk的根目錄kafka相關節點
ls /brokers/ids    #查看kafka節點

# 停止kafka
bin/kafka-server-stop.sh


第五步 命令測試消息
1、創建主題
bin/kafka-topics.sh --create --zookeeper 192.168.31.239:2181 --replication-factor 1 --partitions 1 --topic test
2、列出所有主題
bin/kafka-topics.sh --list --zookeeper 192.168.31.239:2181

3、發送消息
bin/kafka-console-producer.sh --broker-list 192.168.31.243:9092 --topic test

4、消費消息，默認是消費最新的消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.31.243:9092 --topic test

5、創建多個分區主題
bin/kafka-topics.sh --create --zookeeper 192.168.31.239:2181 --replication-factor 1 --partitions 2 --topic test1
擴容分區
bin/kafka-topics.sh -alter --partitions 3 --zookeeper 192.168.31.239:2181 --topic test1

6、查看topic情況
bin/kafka-topics.sh --describe --zookeeper 192.168.31.239:2181 --topic test1

7、消費者的消費偏移量是消費者自己維護的，查看主題的消費偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.31.243:9092 --describe --group test



kafka集群的搭建：
一、集群服務器
zookeeper 1台：192.168.31.239
kafka 2台： 192.168.31.243   192.168.31.244
1、修改kafka配置文件
在192.168.31.243服務器上面
vim config/server.properties

#broker.id屬性在kafka集群中必須要是唯一
broker.id=0
#kafka部署的機器ip和提供服務的端口號
listeners=PLAINTEXT://192.168.31.243:9093   
log.dir=/usr/local/data/kafka-logs-1
#kafka連接zookeeper的地址，要把多個kafka實例組成集群，對應連接的zookeeper必須相同
zookeeper.connect=192.168.31.239:2181

在192.168.31.244服務器上面
vim config/server.properties

#broker.id屬性在kafka集群中必須要是唯一
broker.id=1
#kafka部署的機器ip和提供服務的端口號
listeners=PLAINTEXT://192.168.31.244:9093   
log.dir=/usr/local/data/kafka-logs-1
#kafka連接zookeeper的地址，要把多個kafka實例組成集群，對應連接的zookeeper必須相同
zookeeper.connect=192.168.31.239:2181


2、分別啟動兩台kafka
bin/kafka-server-start.sh -daemon config/server.properties

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

3、測試
 創建一個新的topic，副本數設置為3，分區數設置為2
bin/kafka-topics.sh --create --zookeeper 192.168.31.239:2181 --replication-factor 2 --partitions 2 --topic my-topic
 查看topic信息
 bin/kafka-topics.sh --describe --zookeeper 192.168.31.239:2181 --topic my-topic
 向my-topic主題發送消息
bin/kafka-console-producer.sh --broker-list 192.168.31.243:9092,192.168.31.244:9092 --topic my-topic
消費消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.31.243:9092,192.168.31.244:9092 --from-beginning --topic my-topic