Ubuntu Kafka消費者配置指南

在Ubuntu上配置Kafka消費者涉及多個步驟，包括安裝必要的軟件、配置Kafka和Zookeeper、創建和訂閱主題，以及編寫和運行消費者代碼。以下是一個詳細的指南：

1. 安裝Java和Zookeeper

首先，確保你的系統上安裝了Java和Zookeeper。

安裝Java

sudo apt update
sudo apt install openjdk-8-jdk
java -version

安裝Zookeeper

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz
tar xvf zookeeper-3.4.6.tar.gz
sudo mv zookeeper-3.4.6 /usr/local/zookeeper
sudo cat > /usr/local/zookeeper/conf/zoo.cfg << EOF
tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
EOF
sudo /usr/local/zookeeper/bin/zkServer.sh start

2. 安裝Kafka

接下來，下載并安裝Kafka。

下載Kafka

wget https://downloads.apache.org/kafka/3.5.2/kafka_2.12-3.5.2.tgz
tar -xzvf kafka_2.12-3.5.2.tgz
sudo mv kafka_2.12-3.5.2 /usr/local/kafka
sudo mkdir /tmp/kafka-logs

配置Kafka

編輯Kafka的配置文件server.properties：

sudo nano /usr/local/kafka/config/server.properties

確保以下配置正確：

listeners=PLAINTEXT://your_server_ip:9092
zookeeper.connect=localhost:2181

啟動Kafka

sudo /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

3. 創建和訂閱主題

使用Kafka命令行工具創建一個主題并訂閱它。

創建主題

sudo /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

訂閱主題

sudo /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

4. 編寫和運行消費者代碼

你可以使用Python編寫一個簡單的Kafka消費者。

安裝Kafka-Python

sudo pip3 install kafka-python

消費者代碼示例 (consumer.py)

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=False
)

for msg in consumer:
    print(msg.value.decode('utf-8'))

運行消費者

python3 consumer.py

5. 高級配置

消費者屬性

• auto.offset.reset: 控制消費者在沒有有效偏移量或分區不存在時從哪里開始讀取。
• enable.auto.commit: 是否自動提交偏移量。
• max.poll.records: 每次poll調用返回的最大記錄數。
• fetch.min.bytes: 每次fetch請求的最小字節數。
• max.partition.fetch.bytes: 每個分區返回給消費者的最大字節數。
• session.timeout.ms: 消費者發送心跳信號的時間間隔。

6. 開機自啟配置

為了確保Kafka和Zookeeper在系統啟動時自動啟動，可以配置它們的開機自啟。

Zookeeper服務配置

sudo nano /lib/systemd/system/zookeeper.service

添加內容：

[Unit]
Description=Zookeeper service
After=network.target

[Service]
Type=simple
Environment="PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/local/kafka/bin"
User=root
Group=root
ExecStart=/usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties
ExecStop=/usr/local/kafka/bin/zookeeper-server-stop.sh
Restart=on-failure

[Install]
WantedBy=multi-user.target

Kafka服務配置

sudo nano /lib/systemd/system/kafka.service

添加內容：

[Unit]
Description=Apache Kafka server (broker)
After=zookeeper.service

[Service]
Type=simple
Environment="PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/local/kafka/bin"
User=root
Group=root
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh
Restart=on-failure

[Install]
WantedBy=multi-user.target

刷新配置并啟動服務

sudo systemctl daemon-reload
sudo systemctl enable zookeeper
sudo systemctl enable kafka
sudo systemctl start zookeeper
sudo systemctl start kafka

通過以上步驟，你可以在Ubuntu上成功配置Kafka消費者。根據具體需求，你可以進一步調整消費者配置以滿足實際應用場景。