開發手冊

# 前言

本文檔為金蝶Apusic分散式訊息佇列for Kafka（Apusic Distributed Message Queue，簡稱：ADMQ for Kafka）的開發使用說明，幫助使用者快速學習如何使用金蝶Apusic分散式訊息佇列for Kafka進行開發。

# 適用對象

本文檔適用於IT資訊化業務負責人、研發經理、軟體專案經理、軟體架構師、運維工程師。

# 相關文件

了解更多ADMQ for Kafka產品相關的資訊，請參閱以下ADMQ for Kafka產品手冊文件集：

序號 手冊文件 說明
1 金蝶Apusic分散式訊息佇列for Kafka 快速使用手冊 簡單介紹了如何快速上手使用ADMQ for Kafka。
2 金蝶Apusic分散式訊息佇列for Kafka 安裝手冊 詳細介紹如何在各作業系統上安裝ADMQ for Kafka，以及ADMQ for Kafka服務啟停等操作。
3 金蝶Apusic分散式訊息佇列for Kafka 訊息引擎使用者手冊 詳細介紹ADMQ for Kafka訊息引擎相關功能的使用、設定、管理及配套工具的使用方法。
4 金蝶Apusic分散式訊息佇列for Kafka 管控台使用者手冊 詳細介紹ADMQ for Kafka管控台相關功能的使用和操作說明。
5 金蝶Apusic分散式訊息佇列for Kafka 開發手冊 詳細介紹基於各開發語言進行ADMQ for Kafka用戶端應用開發的說明。
6 金蝶Apusic分散式訊息佇列for Kafka 遷移手冊 詳細介紹從Kafka遷移到ADMQ for Kafka的說明。
7 金蝶Apusic分散式訊息佇列for Kafka 運維手冊 詳細介紹ADMQ for Kafka的監控、運維、安全加固等運維說明。
8 金蝶Apusic分散式訊息佇列for Kafka 效能最佳化手冊 詳細介紹ADMQ for Kafka效能調優的說明。

# 技術支援

ADMQ for Kafka產品提供全面的技術支援服務，您可以透過以下方式獲得技術支援：

  • 網址：www.apusic.com
  • 電話：400-855-5800
  • 信箱：support@apusic.com
  • 金蝶雲社群：https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

您在取得技術支援時，請提供如下資訊：

  1. 您的姓名

  2. 公司資訊與聯絡方式

  3. 作業系統及其版本

  4. 產品版本號

  5. 出現異常及錯誤的日誌、截圖等詳細資訊

# 核心架構

# 主題與分割區

  • Topic（主題）: 訊息的邏輯分類
  • Partition（分割區）: Topic 的物理分片，實現平行讀寫與水平擴充
    • 單個分割區內的訊息嚴格有序
    • 不同分割區之間無全域性順序保證
    • 分割區數應根據吞吐量需求設定，典型值為 10-50

# 副本機制

  • Leader 副本: 處理所有讀寫請求
  • Follower 副本: 僅從 Leader 同步資料，不處理用戶端請求
  • ISR（In-Sync Replicas）: 與 Leader 保持同步的副本集合
    • 只有 ISR 中的副本才有資格成為新 Leader
    • 透過 replica.lag.time.max.ms 控制同步超時（預設 30 秒）

# 高可用與故障轉移

  • Controller 監控 Broker 狀態

  • Leader 故障時，從 ISR 中選舉新 Leader（優先選擇偏移量最高的副本）

  • 更新叢集中繼資料，用戶端自動重新連線新 Leader

  • 預設禁止非 ISR 副本成為 Leader（unclean.leader.election.enable=false）

# 安裝部署

# 單機部署

解壓安裝包

mkdir /apusic
tar -zxf ADMQ-V2.0.6.391-Kafka-20260319.tar.gz -C /apusic
1
2

修改配置

cd /apusic/admq-kafka
vi config/kafka-standalone.conf
...
# 下面位址改成伺服器實際 IP
listeners=PLAINTEXT://192.168.1.10:9092
...
1
2
3
4
5
6

啟動

bin/admq-daemon start kafka zk 
bin/admq-daemon start kafka standalone
1
2

# 用戶端開發

# Maven 依賴

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.1</version>
</dependency>
1
2
3
4
5

# 生產者（Producer）核心配置

配置項 推薦值 說明
bootstrap.servers host1:9092,host2:9092 叢集位址清單
acks all 等待所有 ISR 副本確認
enable.idempotence true 啟用冪等性，防止重複
compression.type lz4 或 zstd 壓縮演算法
linger.ms 5-10 批次傳送延遲
batch.size 16384 批次大小（位元組）
retries Integer.MAX_VALUE 重試次數
max.in.flight.requests.per.connection 5 單連線未確認請求數（冪等性啟用時 ≤ 5）

# 生產者範例程式碼

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "value1");
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println("Sent to partition " + metadata.partition() + 
                          ", offset " + metadata.offset());
    }
});

producer.close();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 消費者（Consumer）核心配置

配置項 推薦值 說明
bootstrap.servers host1:9092,host2:9092 叢集位址
group.id 自訂 消費者組 ID
enable.auto.commit false 停用自動提交，手動控制
auto.offset.reset earliest 或 latest 無位移時策略
max.poll.records 500 單次 poll 最大記錄數
session.timeout.ms 30000 工作階段超時
heartbeat.interval.ms 10000 心跳間隔（應小於工作階段超時的 1/3）

# 消費者範例程式碼

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Received: " + record.value());
        }
        consumer.commitSync();
    }
} finally {
    consumer.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 生產環境最佳實踐

# Topic 設計

  • 分割區數計算：單分割區吞吐 × 分割區數 ≥ 目標吞吐
  • 副本數：生產環境推薦 3 副本
  • 清理策略：
    • cleanup.policy=delete：按時間/大小清理
    • cleanup.policy=compact：日誌壓縮，保留最新 Key

# 效能調優

  • 生產者側：
    • 啟用壓縮（lz4 或 zstd）
    • 合理設定 linger.ms 和 batch.size
    • 啟用冪等性和事務（Exactly-Once 語義）
  • 消費者側：
    • 調整 max.poll.records 避免單次處理過慢
    • 合理設定 fetch.min.bytes 和 fetch.max.wait.ms

# 監控指標

指標 說明 告警閾值
UnderReplicatedPartitions 未充分同步的分割區數 > 0
OfflinePartitionsCount 離線分割區數 > 0
RequestLatencyAvg 平均請求延遲 > 100ms
MessagesInPerSec 每秒訊息數 根據業務設定
BytesInPerSec 每秒寫入位元組數 根據業務設定

# 安全配置

# SASL/PLAIN 認證配置

# broker 配置
listeners=SASL_PLAINTEXT://0.0.0.0:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false

# JAAS 配置檔（以下密碼僅為範例，生產環境請使用強密碼）
KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="admin-secret"
   user_admin="admin-secret"
   user_alice="alice-secret";
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# ACL 範例

# 授予 alice 對 topic my-topic 的讀寫許可權
kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:alice \
  --operation Read --operation Write \
  --topic my-topic

# 授予 alice 對消費者組 my-group 的操作許可權
kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:alice \
  --operation Read --group my-group
1
2
3
4
5
6
7
8
9
10

# 常用命令

ADMQ for Kafka提供原生命令，位於admq-kafka/kafka/bin/目錄下

# Topic 管理

cd /apusic/admq-kafka/kafka/bin/

# 建立 topic
kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic my-topic --partitions 3 --replication-factor 3

# 列出所有 topic
kafka-topics.sh --bootstrap-server localhost:9092 --list

# 檢視 topic 詳情
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic

# 修改分割區數（只能增加）
kafka-topics.sh --bootstrap-server localhost:9092 --alter \
  --topic my-topic --partitions 6

# 刪除 topic
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my-topic
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 消費者組管理

# 列出消費者組
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 檢視消費者組詳情
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group my-group

# 重置位移
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group --reset-offsets --to-earliest \
  --topic my-topic --execute
1
2
3
4
5
6
7
8
9
10
11

# KRaft 模式相關

# 格式化儲存目錄（僅首次）
kafka-storage.sh format -t <cluster-id> -c config/kraft/server.properties

# 檢視仲裁狀態
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

# 動態新增控制器節點
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 add \
  --controller-quorum-voter <node-id>@<host>:<port>
1
2
3
4
5
6
7
8
9

# 故障排查

# 常見問題

問題 可能原因 解決方案
消費者頻繁重新平衡 session.timeout.ms 過小或處理時間過長 增大工作階段超時，最佳化消費邏輯
訊息遺失 acks=1 或未啟用冪等性 設定 acks=all，啟用 enable.idempotence
重複消費 位移提交失敗 使用手動提交，處理重複邏輯
生產延遲高 網路或磁碟瓶頸 檢查網路頻寬，最佳化磁碟 I/O
分割區不可用 Broker 故障或副本不足 檢查 Broker 狀態，確保 ISR 足夠

# 日誌位置

  • Broker 日誌：logs/server.log
  • Controller 日誌：logs/controller.log
  • 狀態變更日誌：logs/state-change.log
编辑页面 (opens new window)
#开发手册

← 遷移手冊→