跳到主要内容

Kafka 集群部署与集成指南

概述​

Apache Kafka 是一个分布式流处理平台，提供高吞吐量的实时数据管道。广泛应用于日志收集、流式处理、事件溯源等场景。

核心概念​

  • 📝 Producer - 消息生产者
  • 📥 Consumer - 消息消费者
  • 📋 Topic - 消息主题（分类）
  • 🔗 Partition - 分区（并行处理单元）
  • 🐘 Broker - Kafka 服务器节点
  • 🏛️ ZooKeeper - 协调服务（Kafka 2.8+ 支持 KRaft 模式可不用 ZK）

架构设计​

┌─────────────┐      ┌─────────────┐
│  Producer   │ ──►  │   Broker 1  │ (Leader: Topic-A-P0)
└─────────────┘      └─────────────┘
                            │
                            ▼
                     ┌─────────────┐
                     │   Broker 2  │ (Follower: Topic-A-P0)
                     └─────────────┘
                            │
                            ▼
                     ┌─────────────┐
                     │  Consumer   │
                     └─────────────┘

集群部署方案​

环境规划​

节点IPBroker IDZooKeeper说明
kafka-1192.168.192.1501✓Controller + Broker
kafka-2192.168.192.1512✓Broker
kafka-3192.168.192.1523✓Broker

前置要求​

  • ✅ ZooKeeper 集群（3 节点）
  • ✅ JDK 11+
  • ✅ 磁盘：每节点 100GB+ SSD
  • ✅ 网络：千兆网卡

安装步骤​

1. 下载 Kafka​

cd /tmp
wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.13-3.6.0.tgz
tar -xzf kafka_2.13-3.6.0.tgz
sudo mv kafka_2.13-3.6.0 /opt/kafka
sudo chown -R kafka:kafka /opt/kafka

2. 配置 ZooKeeper（如果已有 ZK 集群可跳过）​

# 创建数据目录
sudo mkdir -p /var/lib/zookeeper
sudo chown kafka:kafka /var/lib/zookeeper

# 配置 myid
echo "1" | sudo tee /var/lib/zookeeper/myid  # kafka-1 上执行

3. 配置 Kafka​

编辑 /opt/kafka/config/server.properties：

# Broker 唯一标识
broker.id=1

# 监听地址
listeners=PLAINTEXT://192.168.192.150:9092
advertised.listeners=PLAINTEXT://192.168.192.150:9092

# 日志目录
log.dirs=/var/lib/kafka/logs

# ZooKeeper 连接
zookeeper.connect=192.168.192.150:2181,192.168.192.151:2181,192.168.192.152:2181

# 分区数（默认）
num.partitions=3

# 副本数
default.replication.factor=3
min.insync.replicas=2

4. 启动服务​

# 启动 ZooKeeper（如果没有独立 ZK 集群）
/opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

# 启动 Kafka
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

与 DolphinScheduler 集成​

在 DS 中使用 Kafka​

1. Python 任务生产消息​

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['192.168.192.150:9092', '192.168.192.151:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 发送消息
message = {'event': 'order_created', 'order_id': 12345}
producer.send('orders_topic', value=message)
producer.flush()
producer.close()

print("消息发送成功")

2. Shell 任务消费消息​

#!/bin/bash

# 消费 Kafka 消息
/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.150:9092 \
  --topic orders_topic \
  --from-beginning \
  --max-messages 100

3. SeaTunnel 读取 Kafka​

source {
  Kafka {
    bootstrap.servers = "192.168.192.150:9092,192.168.192.151:9092"
    topic = "orders_topic"
    format = "json"
  }
}

sink {
  Doris {
    fenodes = "192.168.192.140:8030"
    database = "realtime_db"
    table = "orders"
  }
}

典型应用场景​

场景一：实时日志收集​

App Server → Filebeat → Kafka → Logstash → Elasticsearch → Kibana

场景二：事件驱动架构​

订单服务 → Kafka → 
  ├─ 库存服务
  ├─ 支付服务
  └─ 通知服务

场景三：实时数据管道​

MySQL (CDC) → Kafka → 
  ├─ Doris (实时数仓)
  ├─ HDFS (数据湖)
  └─ Redis (缓存)

运维管理​

创建 Topic​

# 创建主题
/opt/kafka/bin/kafka-topics.sh \
  --create \
  --bootstrap-server 192.168.192.150:9092 \
  --replication-factor 3 \
  --partitions 6 \
  --topic orders_topic

查看 Topic 信息​

# 列出所有主题
/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server 192.168.192.150:9092

# 查看详情
/opt/kafka/bin/kafka-topics.sh \
  --describe \
  --topic orders_topic \
  --bootstrap-server 192.168.192.150:9092

监控指标​

  • Throughput: 消息生产/消费速率
  • Lag: 消费延迟
  • Under Replicated Partitions: 未同步分区数
  • Active Controller Count: 活跃的 Controller 数量

性能优化​

# Producer 优化
batch.size=16384
linger.ms=5
compression.type=lz4

# Consumer 优化
fetch.min.bytes=1
fetch.max.wait.ms=500

最佳实践​

1. 高可用配置​

# Broker 配置
min.insync.replicas=2
unclean.leader.election.enable=false
auto.leader.rebalance.enable=true

2. 数据保留策略​

# 基于时间
retention.ms=604800000  # 7 天

# 基于大小
retention.bytes=10737418240  # 10GB

3. 安全配置​

# 启用 SSL
security.inter.broker.protocol=SSL
ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks
ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks

常见问题​

Q1: Kafka 需要 ZooKeeper 吗？​

答：

  • Kafka 2.8 之前: 必须依赖 ZooKeeper
  • Kafka 2.8+: 支持 KRaft 模式（无需 ZK）
  • Kafka 3.0+: KRaft 模式成熟，推荐生产使用

Q2: 如何保证消息不丢失？​

# Producer
acks=all
retries=3
enable.idempotence=true

# Broker
min.insync.replicas=2
unclean.leader.election.enable=false

# Consumer
enable.auto.commit=false

Q3: 如何处理消息积压？​

  1. 增加 Consumer 数量
  2. 增加 Topic 分区数
  3. 优化 Consumer 处理逻辑
  4. 临时扩容 Consumer 集群

参考资源​