kafka集群部署與驗證

kafka集群部署與驗證

一、官網下載kafka

wget https://www.apache.org/dyn/closer.cgi?path=/kafka/2.3.0/kafka_2.12-2.3.0.tgz

二、zookeeper

2.1、配置zookeeper

tar -zxvf kafka_2.12-2.3.0.tgz
cd /opt/kafka_2.12-2.3.0/config

[miaocunfa@db1 config]$ cat zookeeper.properties | grep -v ^# | grep -v ^$

# tickTime：心跳基本時間單位，毫秒級，ZK基本上所有的時間都是這個時間的整數倍。
# initLimit：tickTime的個數，表示在leader選舉結束后，followers與leader同步需要的時間，如果followers比較多或者說leader的數據灰常多時，同步時間相應可能會增加，那么這個值也需要相應增加。當然，這個值也是follower和observer在開始同步leader的數據時的最大等待時間(setSoTimeout)
# syncLimit：tickTime的個數，這時間容易和上面的時間混淆，它也表示follower和observer與leader交互時的最大等待時間，只不過是在與leader同步完畢之后，進入正常請求轉發或ping等消息交互時的超時時間。
tickTime=2000
initLimit=10
syncLimit=5

# 內存數據庫快照存放地址，如果沒有指定事務日志存放地址(dataLogDir)，默認也是存放在這個路徑下，建議兩個地址分開存放到不同的設備上。
dataDir=/ahdata/kafka-tmp/zookeeper

# 配置ZK監聽客戶端連接的端口
clientPort=2181

# 默認值是10，一個客戶端能夠連接到同一個服務器上的最大連接數，根據IP來區分。如果設置為0，表示沒有任何限制。設置該值一方面是為了防止DoS***。
maxClientCnxns=0

# server.serverid=host:tickpot:electionport
# server：固定寫法
# serverid：每個服務器的指定ID（必須處于1-255之間，必須每一臺機器不能重復）
# host：主機名
# tickpot：心跳通信端口
# electionport：選舉端口
server.1=172.19.26.3:2888:3888
server.2=172.19.26.6:2888:3888
server.3=172.19.26.4:2888:3888

2.2 各節點分別創建server-id

# myid文件創建在dataDir目錄下
# myid內容與配置文件中的serverid一致
echo 1 > /ahdata/kafka-tmp/zookeeper/myid
echo 2 > /ahdata/kafka-tmp/zookeeper/myid
echo 2 > /ahdata/kafka-tmp/zookeeper/myid

2.3、啟動zookeeper

/opt/kafka_2.12-2.3.0/bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
/opt/kafka_2.12-2.3.0/bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
/opt/kafka_2.12-2.3.0/bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

2.4、驗證zookeeper狀態

[root@db1 config]# echo stat | nc 172.19.26.3 2181 | grep Mode
Mode: follower
[root@db1 config]# echo stat | nc 172.19.26.4 2181 | grep Mode
Mode: follower
[root@db1 config]# echo stat | nc 172.19.26.6 2181 | grep Mode
Mode: leader

三、kafka

3.1、配置kafka

# 當前機器在集群中的唯一標識，和zookeeper的myid性質一樣
broker.id=1

# 學名叫監聽器，其實就是告訴外部連接者要通過什么協議訪問指定主機名和端口開放的 Kafka 服務。
listeners=PLAINTEXT://172.19.26.3:9092
# listeners 相比多了個 advertised。Advertised 的含義表示宣稱的、公布的，就是說這組監聽器是 Broker 用于對外發布的。
advertised.listeners=PLAINTEXT://172.19.26.3:9092

# broker 處理消息的最大線程數，一般情況下不需要去修改
num.network.threads=3
# broker處理磁盤IO 的線程數 ，數值應該大于你的硬盤數
num.io.threads=8

# 發送緩沖區buffer大小，數據不是一下子就發送的，先回存儲到緩沖區了到達一定的大小后在發送，能提高性能
socket.send.buffer.bytes=102400
# kafka接收緩沖區大小，當數據到達一定大小后在序列化到磁盤
socket.receive.buffer.bytes=102400
# 這個參數是向kafka請求消息或者向kafka發送消息的請請求的最大數，這個值不能超過java的堆棧大小
socket.request.max.bytes=104857600

# 如果配置多個目錄，新創建的topic他把消息持久化的地方是，當前以逗號分割的目錄中，那個分區數最少就放那一個
log.dirs=/ahdata/kafka-tmp/kafka-logs

# 分區數，一個topic 3個分區
num.partitions=3

# 每個數據目錄用來日志恢復的線程數目
num.recovery.threads.per.data.dir=1

# 集群高可用參數，建議使用大于1的值來確保可用性，比如3。
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3

# 默認消息的最大持久化時間，168小時，7天
log.retention.hours=168
#這個參數是：因為kafka的消息是以追加的形式落地到文件，當超過這個值的時候，kafka會新起一個文件
log.segment.bytes=1073741824
# 每隔300000毫秒去檢查上面配置的log失效時間
log.retention.check.interval.ms=300000

# zookeeper的連接端口
zookeeper.connect=172.19.26.3:2181,172.19.26.4:2181,172.19.26.6:2181
# zookeeper的連接超時時間
zookeeper.connection.timeout.ms=6000

# 客戶端消費者重新選舉的延時時間，默認0
group.initial.rebalance.delay.ms=0

# 允許刪除topic
delete.topic.enable=true

3.2、啟動kafka

/opt/kafka_2.12-2.3.0/bin/kafka-server-start.sh -daemon config/server.properties
/opt/kafka_2.12-2.3.0/bin/kafka-server-start.sh -daemon config/server.properties
/opt/kafka_2.12-2.3.0/bin/kafka-server-start.sh -daemon config/server.properties

3.3、驗證kafka狀態

[root@db1 kafka_2.12-2.3.0]# echo dump | nc 172.19.26.3 2181 | grep broker
    /brokers/ids/1
    /brokers/ids/2
[root@db1 kafka_2.12-2.3.0]# echo dump | nc 172.19.26.4 2181 | grep broker
    /brokers/ids/1
    /brokers/ids/2
[root@db1 kafka_2.12-2.3.0]# echo dump | nc 172.19.26.6 2181 | grep broker
    /brokers/ids/1
    /brokers/ids/2

四、驗證集群

4.1、topic

# 創建一個topic
[miaocunfa@db1 kafka_2.12-2.3.0]$ bin/kafka-topics.sh --create --zookeeper 172.19.26.3:2181,172.19.26.4:2181,172.19.26.6:2181 --replication-factor 2 --partitions 3 --topic demo_topics
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic demo_topics.

# 列出所有topic
[miaocunfa@db1 kafka_2.12-2.3.0]$ /opt/kafka_2.12-2.3.0/bin/kafka-topics.sh --list --zookeeper 172.19.26.3:2181,172.19.26.4:2181,172.19.26.6:2181
demo_topics

# 查看topic詳細情況
[miaocunfa@db1 config]$ /opt/kafka_2.12-2.3.0/bin/kafka-topics.sh --describe --zookeeper 172.19.26.3:2181,172.19.26.4:2181,172.19.26.6:2181 --topic demo_topics
Topic:demo_topics   PartitionCount:3    ReplicationFactor:2 Configs:
    Topic: demo_topics  Partition: 0    Leader: 2   Replicas: 1,2   Isr: 2,1
    Topic: demo_topics  Partition: 1    Leader: 2   Replicas: 2,1   Isr: 2,1
    Topic: demo_topics  Partition: 2    Leader: 2   Replicas: 1,2   Isr: 2,1

4.2、生產消費驗證

ps. 1) 若producer 和 consumer 兩個窗口同時打開，在producer輸入信息，consumer會立即消費信息并打印在終端
    2）新開一個終端，去消費同一個topic，剛剛已經消費過的消息還會被新終端繼續消費。也就是說，消息被消費過后不會立即被刪除。　

4.2.1、生產者發送消息

[miaocunfa@db1 kafka_2.12-2.3.0]$ /opt/kafka_2.12-2.3.0/bin/kafka-console-producer.sh --broker-list 172.19.26.3:9092,172.19.26.4:9092,172.19.26.6:9092 --topic demo_topics
>Hello Kafka!      
>

4.2.2、消費者接收消息

# 啟動一個新終端創建一個消費者接收消息。
[root@db1 kafka_2.12-2.3.0]# /opt/kafka_2.12-2.3.0/bin/kafka-console-consumer.sh --bootstrap-server=172.19.26.3:9092,172.19.26.4:9092,172.19.26.6:9092 --topic demo_topics --from-beginning
Hello Kafka!

4.3、刪除測試topic

# 配置文件中delete.topic.enable=true才可刪除topic
[miaocunfa@db1 config]$ /opt/kafka_2.12-2.3.0/bin/kafka-topics.sh --delete --zookeeper 172.19.26.3:2181,172.19.26.4:2181,172.19.26.6:2181 --topic demo_topics
Topic demo_topics is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

參考：
1.https://www.cnblogs.com/qingyunzong/p/8619184.html
2.https://www.cnblogs.com/qingyunzong/p/9005062.html#_label3_5
3.https://www.cnblogs.com/cici20166/p/9426417.html
4.https://www.orchome.com/805