# Kafka如何安裝并實現單機測試
## 一、Kafka簡介
Apache Kafka是由LinkedIn開發并開源的高性能分布式消息系統,具有以下核心特性:
- **高吞吐量**:單機可支持每秒百萬級消息處理
- **持久化存儲**:消息可持久化到磁盤并支持多副本
- **分布式架構**:天然支持水平擴展
- **低延遲**:消息投遞延遲可控制在毫秒級
- **高容錯性**:支持自動故障轉移
### 1.1 核心組件
| 組件 | 說明 |
|---------------|----------------------------------------------------------------------|
| Producer | 消息生產者,負責發布消息到指定Topic |
| Consumer | 消息消費者,訂閱Topic并處理消息 |
| Broker | Kafka服務實例,負責消息存儲和轉發 |
| Topic | 消息類別/主題,邏輯上的消息分類 |
| Partition | Topic的物理分片,每個Partition是一個有序、不可變的消息隊列 |
| Zookeeper | 負責集群元數據管理、Broker選舉等協調工作(Kafka 2.8+開始支持去ZK模式)|
## 二、單機環境安裝
### 2.1 環境準備
**系統要求**:
- 推薦Linux/MacOS(Windows可能有兼容性問題)
- JDK 1.8+(建議OpenJDK 11)
- 至少4GB可用內存
- 10GB以上磁盤空間
```bash
# 檢查Java環境
java -version
wget https://downloads.apache.org/kafka/3.3.1/kafka_2.13-3.3.1.tgz
tar -xzf kafka_2.13-3.3.1.tgz
cd kafka_2.13-3.3.1
目錄結構說明:
bin/ # 操作腳本
config/ # 配置文件
libs/ # 依賴庫
logs/ # 日志文件(啟動后生成)
修改config/server.properties
核心參數:
# Broker唯一標識
broker.id=0
# 監聽地址
listeners=PLNTEXT://localhost:9092
# 日志存儲目錄
log.dirs=/tmp/kafka-logs
# ZooKeeper連接地址
zookeeper.connect=localhost:2181
# 自動創建Topic(測試環境建議開啟)
auto.create.topics.enable=true
Kafka依賴ZooKeeper,單機版可使用內置ZK:
# 后臺啟動ZooKeeper
nohup bin/zookeeper-server-start.sh config/zookeeper.properties > zk.log 2>&1 &
# 驗證啟動
ps aux | grep zookeeper
# 啟動Kafka服務
nohup bin/kafka-server-start.sh config/server.properties > kafka.log 2>&1 &
# 檢查是否啟動成功
tail -f logs/server.log
成功日志示例:
[KafkaServer id=0] started (kafka.server.KafkaServer)
# 創建Topic(1分區1副本)
bin/kafka-topics.sh --create \
--bootstrap-server localhost:9092 \
--replication-factor 1 \
--partitions 1 \
--topic test-topic
# 查看Topic列表
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# 查看Topic詳情
bin/kafka-topics.sh --describe \
--topic test-topic \
--bootstrap-server localhost:9092
生產者發送消息:
bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 \
--topic test-topic
> Hello Kafka
> This is a test message
消費者接收消息:
bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic test-topic \
--from-beginning
內置壓測工具:
# 生產者性能測試
bin/kafka-producer-perf-test.sh \
--topic perf-test \
--num-records 100000 \
--record-size 1000 \
--throughput 2000 \
--producer-props bootstrap.servers=localhost:9092
# 消費者性能測試
bin/kafka-consumer-perf-test.sh \
--topic perf-test \
--bootstrap-server localhost:9092 \
--messages 100000
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.3.1</version>
</dependency>
public class SimpleProducer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 10; i++) {
producer.send(new ProducerRecord<>("test-topic",
Integer.toString(i),
"Message-" + i));
}
producer.close();
}
}
public class SimpleConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test-topic"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset=%d, key=%s, value=%s%n",
record.offset(), record.key(), record.value());
}
}
}
}
端口沖突:
netstat -tulnp | grep 9092
ZooKeeper連接問題:
zookeeper.connect
配置tail -f zk.log
磁盤空間不足:
df -h
acks
配置(0/1/all)group.id
是否相同(相同group會分片消費)auto.offset.reset
配置(earliest/latest/none)參數 | 建議值 | 說明 |
---|---|---|
num.network.threads | 3 | 網絡線程數 |
num.io.threads | 8 | IO線程數(建議>=磁盤數) |
socket.send.buffer.bytes | 1024000 | 發送緩沖區大小 |
socket.receive.buffer.bytes | 1024000 | 接收緩沖區大小 |
log.retention.hours | 168 | 日志保留時間(小時) |
JMX監控:
export JMX_PORT=9999
bin/kafka-server-start.sh config/server.properties
Kafka Eagle可視化工具:
docker pull smartloli/kafka-eagle
本文詳細介紹了Kafka單機環境的: 1. 安裝與配置步驟 2. 基礎功能驗證方法 3. Java客戶端開發示例 4. 常見問題解決方案
后續可進一步探索: - Kafka Connect數據集成 - Kafka Streams流處理 - KRaft模式(去ZooKeeper化部署)
注意事項:生產環境需配置多副本、監控告警等機制,本文單機配置僅適用于開發和測試場景。 “`
該文檔共計約2300字,包含: - 安裝部署詳細步驟 - 配置說明與性能測試 - Java客戶端示例代碼 - 常見問題解決方案 - 格式規范的Markdown排版
可通過實際執行文中命令快速搭建可用的Kafka測試環境。
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。