kafka環境搭建

測驗環境

centos 7.8.2003

JDK安裝

下載地址： https://www.oracle.com/java/technologies/downloads/

當前下載路徑：https://download.oracle.com/java/17/latest/jdk-17_linux-x64_bin.tar.gz

下載安裝：

$ curl -O https://download.oracle.com/java/17/latest/jdk-17_linux-x64_bin.tar.gz
$ tar -xvf jdk-17_linux-x64_bin.tar.gz
#jdk-17.0.1
$ echo "export JAVA_HOME=`pwd`/jdk-17.0.1" >>/etc/profile
$ echo 'export CLASSPATH=$:CLASSPATH:$JAVA_HOME/lib/' >>/etc/profile
$ echo 'export PATH=$PATH:$JAVA_HOME/bin' >>/etc/profile
$ source /etc/profile
$ java -version
java version "17.0.1" 2021-10-19 LTS
Java(TM) SE Runtime Environment (build 17.0.1+12-LTS-39)
Java HotSpot(TM) 64-Bit Server VM (build 17.0.1+12-LTS-39, mixed mode, sharing)

單節點

下載安裝

#下載地址：http://kafka.apache.org/downloads
#當前版本：2.8.1
$ curl -O https://archive.apache.org/dist/kafka/2.8.1/kafka_2.13-2.8.1.tgz
$ tar -xvf kafka_2.13-2.8.1.tgz
#進入kafka目錄
$ cd kafka_2.13-2.8.1

組態檔修改

$ vi config/server.properties 
	#添加監聽的ip埠配置
	listeners=PLAINTEXT://<ip>:9092
	#日志存盤目錄
	log.dirs=<log_dir>/kafka-logs

啟動

#先啟動zookeeper
$ ./bin/zookeeper-server-start.sh ./config/zookeeper.properties
#然后啟動kafka
$ ./bin/kafka-server-start.sh ./config/server.properties

zookeeper查看節點資訊

$ ./bin/zookeeper-shell.sh localhost:2181
#輸入如下命令（id為kafka組態檔中的broker.id）：
#不知道id可以先查看
#跟多命令可以使用help查看
> ls /brokers/ids
[0]
> get /brokers/ids/0
{"features":{},"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://a4b96b9307b2:9092"],"jmx_port":-1,"port":9092,"host":"a4b96b9307b2","version":5,"timestamp":"1634886207746"}
# 查看topic磁區資訊
> ls /brokers/topics/<topic name>/partitions

基于腳本的kafka操作

所有腳本可在下載的bin目錄下找到

官網介紹：https://kafka.apache.org/documentation/#configuration

#主題(topic)
#測驗主題名test
#查看topic串列
$ ./bin/kafka-topics.sh --list --bootstrap-server localhost:9092
#創建topic, --partitions: 磁區數，--replication-factor： 副本數，不能超過broker數量
$ ./bin/kafka-topics.sh --bootstrap-server <ip>:9092 --create --topic test --partitions 20 --replication-factor 3 --config x=y
#修改topic配置和磁區配置
#磁區只能增加，不支持減少
$ ./bin/kafka-topics.sh --bootstrap-server <ip>:9092 --alter --topic test --partitions 30

#修改配置
#增加配置
$ ./bin/kafka-configs.sh --bootstrap-server <ip>:9092 --entity-type topics --entity-name test --alter --add-config x=y
#洗掉配置
$ ./bin/kafka-configs.sh --bootstrap-server <ip>:9092 --entity-type topics --entity-name test --alter --delete-config x
#洗掉主題
$ ./bin/kafka-topics.sh --bootstrap-server <ip>:9092 --delete --topic test

#獲取磁區最新offset
$ ./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <ip>:9092 --topic test

消費測驗

#啟動消費者，消費訊息
./bin/kafka-console-consumer.sh --bootstrap-server <ip>:9092 --topic test1 --group group1
#啟動生產者，生產訊息
./bin/kafka-console-producer.sh --bootstrap-server <ip>:9092 --topic test1

集群

zookeeper使用kafka包里自帶的，集群最少需要三臺服務器，且官方強烈建議使用奇數臺服務器，

zookeeper配置：

更多配置參考官網（版本3.7.0）：https://zookeeper.apache.org/doc/r3.7.0/zookeeperStarted.html

$ vi config/zookeeper.properties
   	#注釋可能不準確，原文解釋可查看官網
    #zookeeper使用的基本時間單位，單位是ms，用于心跳，最小會話超時將是tickTime 的兩倍
    tickTime=2000
    #簡單理解為保存資料的目錄
    dataDir=/var/lib/zookeeper
    #監聽埠
    clientPort=2181
    #連接到主節點的超時時間，使用tickTime為時間單位，即超時時間為2000 * 5 = 10000ms = 10s
    initLimit=5
    #從節點與主節點同步的超時時間，使用tickTime為時間單位，即超時時間為2000 * 2 = 4000ms = 4s
    syncLimit=2
    #集群資訊，第一個ip用于從節點與主節點進行通信，第二個ip用于選舉主節點，都是tcp通信
    server.1=<zoo1>:2888:3888
    server.2=<zoo2>:2888:3888
    server.3=<zoo3>:2888:3888
#創建zookeeper集群需要的myid檔案(檔案路徑為組態檔中的<dataDir>), 然后在里面添加一個數字來標識此節點，與kafka broker id類似
$ echo 1 >> <dataDir>/myid

kafka配置：

$ vi config/server.properties
	#broker id，每個節點需設定不同的值
	broker.id=0
	#添加zookeeper集群資訊
	zookeeper.connect=<zoo1>:2181,<zoo2>:2181,<zoo3>:2181

kafka官方建議需配置項：

# ZooKeeper
zookeeper.connect=[list of ZooKeeper servers]

# Log configuration
num.partitions=8
default.replication.factor=3 # 不能大于集群數量
log.dir=[List of directories. Kafka should have its own dedicated disk(s) or SSD(s).]

# Other configurations
broker.id=[An integer. Start with 0 and increment by 1 for each new broker.]
listeners=[list of listeners]
auto.create.topics.enable=false
min.insync.replicas=2
queued.max.requests=[number of concurrent requests]