Python Kafka 开发利器：confluent-kafka-python 从入门到实战

作者：

在

一、confluent-kafka-python 核心概述

1.1 库的用途

confluent-kafka-python 是 Confluent 公司推出的 Kafka Python 客户端，基于高性能的 librdkafka C 库封装而成，主要用于在 Python 程序中实现与 Apache Kafka 集群的高效交互，支持生产者（Producer）向 Kafka 发送消息、消费者（Consumer）从 Kafka 订阅并消费消息，同时兼容 Kafka 的各种高级特性，广泛应用于实时数据管道、日志收集、消息队列解耦等场景。

1.2 工作原理

该库的底层依赖 librdkafka，这是一个工业级的 Kafka 客户端库，提供了可靠的消息传输机制。在 Python 层面，confluent-kafka-python 对 librdkafka 的 API 进行了轻量级封装，实现了生产者的消息分区策略、批量发送、消息确认，以及消费者的群组协调、自动提交偏移量、消息回溯等核心功能。其工作流程遵循 Kafka 的标准模型：生产者将消息发送到指定 Topic，Kafka 集群存储消息，消费者订阅 Topic 并拉取消息进行处理。

1.3 优缺点分析

优点

性能优异：基于 C 语言的 librdkafka，吞吐量和延迟表现远超纯 Python 实现的 Kafka 客户端（如 kafka-python）。
功能全面：支持 Kafka 的所有核心特性，包括事务消息、压缩算法、SSL 加密、SASL 认证、自定义分区器等。
稳定性高：经过大规模生产环境验证，适合高并发、高可用的场景。
配置灵活：提供丰富的配置参数，可针对生产者和消费者进行精细化调优。

缺点

安装依赖：需要系统中安装 librdkafka 库，Windows 平台安装相对复杂。
学习曲线：部分高级配置参数（如分区策略、偏移量管理）需要对 Kafka 原理有一定理解。
跨平台兼容：在一些小众操作系统上可能存在编译问题，需要手动调整编译参数。

1.4 License 类型

二、confluent-kafka-python 安装与环境准备

2.1 系统依赖安装

由于 confluent-kafka-python 依赖 librdkafka，在安装 Python 包之前需要先安装系统级的 librdkafka 库。

2.1.1 Linux 系统（Ubuntu/Debian）

sudo apt-get update
sudo apt-get install librdkafka-dev

2.1.2 Linux 系统（CentOS/RHEL）

sudo yum install librdkafka-devel

2.1.3 macOS 系统

使用 Homebrew 安装：

brew install librdkafka

2.1.4 Windows 系统

Windows 平台安装相对复杂，推荐两种方式：

使用预编译的二进制包：从 librdkafka 官网下载预编译的 Windows 版本，解压后将库文件路径添加到系统环境变量 PATH 中。
使用 WSL（Windows Subsystem for Linux）：在 WSL 中安装 Linux 版本的依赖，然后在 WSL 中运行 Python 程序。

2.2 Python 包安装

系统依赖安装完成后，使用 pip 安装 confluent-kafka-python：

pip install confluent-kafka

验证安装是否成功：

import confluent_kafka
print(confluent_kafka.__version__)

运行上述代码，如果输出库的版本号（如 2.2.0），则说明安装成功。

三、核心功能实战：生产者与消费者

3.1 Kafka 环境准备

在进行代码实战前，需要确保有一个可用的 Kafka 集群。如果是本地测试，可以使用 Docker 快速启动单节点 Kafka 和 ZooKeeper：

# 启动 ZooKeeper
docker run -d --name zookeeper -p 2181:2181 confluentinc/cp-zookeeper:7.4.0 \
  ZOOKEEPER_CLIENT_PORT=2181 \
  ZOOKEEPER_TICK_TIME=2000

# 启动 Kafka
docker run -d --name kafka -p 9092:9092 --link zookeeper:zookeeper confluentinc/cp-kafka:7.4.0 \
  KAFKA_BROKER_ID=1 \
  KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT \
  KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1

上述命令启动了一个单节点 Kafka 集群，监听本地 9092 端口。

3.2 生产者（Producer）实战

生产者的核心功能是向 Kafka Topic 发送消息。confluent-kafka-python 提供了 Producer 类，支持同步发送、异步发送、批量发送等多种模式。

3.2.1 基础异步生产者

异步发送是 Kafka 生产者的默认模式，特点是无需等待消息发送结果，通过回调函数处理发送成功或失败的通知，效率更高。

代码示例

from confluent_kafka import Producer
import json
import time

# 1. 配置生产者参数
producer_config = {
    "bootstrap.servers": "localhost:9092",  # Kafka 集群地址
    "client.id": "python-producer-demo",    # 客户端标识
    "acks": "1",                            # 消息确认级别：1 表示 leader 确认即可
    "retries": 3,                           # 发送失败重试次数
    "linger.ms": 5,                         # 批量发送延迟时间（毫秒）
    "compression.type": "gzip"              # 消息压缩算法
}

# 2. 初始化生产者
producer = Producer(producer_config)

# 3. 定义发送结果回调函数
def delivery_report(err, msg):
    """
    消息发送结果回调函数
    :param err: 发送失败时的错误信息，成功时为 None
    :param msg: 发送成功的消息元数据
    """
    if err is not None:
        print(f"消息发送失败: {err}")
    else:
        print(f"消息发送成功 -> Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset()}")

# 4. 发送消息
topic = "test_topic"  # 目标 Topic

# 循环发送 10 条测试消息
for i in range(10):
    # 构造消息内容
    message_data = {
        "id": i,
        "content": f"Hello Kafka from Python - {i}",
        "timestamp": time.time()
    }
    # 将字典转换为 JSON 字符串
    message_value = json.dumps(message_data).encode("utf-8")

    # 发送消息：key 用于分区路由，value 为消息内容
    producer.produce(
        topic=topic,
        key=str(i).encode("utf-8"),
        value=message_value,
        on_delivery=delivery_report
    )

    # 触发消息发送（异步模式下需要定期调用 poll 处理事件）
    producer.poll(0)

    # 模拟业务延迟
    time.sleep(0.5)

# 5. 等待所有待发送消息完成
producer.flush()
print("所有消息发送完成！")

代码说明

配置参数：bootstrap.servers 指定 Kafka 集群地址，acks 设置消息确认级别（0=无确认，1=leader 确认，all=所有副本确认），retries 设置重试次数，linger.ms 控制批量发送的延迟时间，compression.type 启用 gzip 压缩以减少网络传输量。
回调函数：delivery_report 函数用于处理消息发送结果，当消息成功发送或失败时会被调用。
produce 方法：用于发送消息，key 会影响消息的分区策略（相同 key 的消息会被发送到同一个分区），value 为消息的二进制内容。
poll 方法：异步模式下必须定期调用 poll 方法，处理 Kafka 的事件（如回调函数执行），参数 0 表示非阻塞。
flush 方法：等待所有待发送的消息完成发送，确保程序退出前消息不会丢失。

3.2.2 同步生产者

同步发送模式下，程序会阻塞直到收到 Kafka 的确认响应，适合对消息发送结果有强依赖的场景。

代码示例

from confluent_kafka import Producer, KafkaError
import json

# 配置生产者参数
producer_config = {
    "bootstrap.servers": "localhost:9092",
    "acks": "all",  # 最高级别确认，确保消息可靠性
    "retries": 5
}

producer = Producer(producer_config)
topic = "test_topic"

def send_message_sync(topic, key, value):
    """
    同步发送消息
    """
    try:
        # 发送消息并等待结果
        producer.produce(topic, key=key, value=value)
        # 阻塞直到消息发送完成
        producer.flush()
        print("消息同步发送成功")
    except KafkaError as e:
        print(f"消息同步发送失败: {e}")

# 构造消息
message_value = json.dumps({"data": "Sync Message from Python"}).encode("utf-8")
send_message_sync(topic, b"sync_key", message_value)

代码说明

同步发送的核心是调用 flush 方法，该方法会阻塞直到所有待发送消息处理完成。
通过捕获 KafkaError 异常，可以处理发送过程中的错误。

3.3 消费者（Consumer）实战

消费者的核心功能是订阅 Kafka Topic 并拉取消息进行处理。confluent-kafka-python 提供了 Consumer 类，支持消费者群组、自动提交偏移量、手动提交偏移量等功能。

3.3.1 基础消费者（自动提交偏移量）

自动提交偏移量是消费者的默认模式，Kafka 会定期自动将消费者的偏移量提交到集群，简化开发流程。

代码示例

from confluent_kafka import Consumer, KafkaError
import json

# 1. 配置消费者参数
consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "python-consumer-group",  # 消费者群组 ID
    "auto.offset.reset": "earliest",      # 当没有初始偏移量时，从最早的消息开始消费
    "enable.auto.commit": True,           # 启用自动提交偏移量
    "auto.commit.interval.ms": 5000       # 自动提交间隔时间（毫秒）
}

# 2. 初始化消费者
consumer = Consumer(consumer_config)

# 3. 订阅 Topic
topic = "test_topic"
consumer.subscribe([topic])
print(f"消费者已订阅 Topic: {topic}")

# 4. 消费消息
try:
    while True:
        # 拉取消息，超时时间设置为 1 秒
        msg = consumer.poll(timeout=1.0)

        # 如果没有消息，继续循环
        if msg is None:
            continue

        # 处理错误
        if msg.error():
            # 处理分区 EOF 事件
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print(f"已到达分区末尾 -> Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset()}")
            else:
                print(f"消费消息出错: {msg.error()}")
            continue

        # 处理正常消息
        key = msg.key().decode("utf-8") if msg.key() else None
        value = json.loads(msg.value().decode("utf-8"))
        print(f"消费到消息 -> Key: {key}, Value: {value}, Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset()}")

except KeyboardInterrupt:
    print("用户中断消费")
finally:
    # 关闭消费者，提交最后一次偏移量
    consumer.close()
    print("消费者已关闭")

代码说明

配置参数：group.id 指定消费者群组 ID，同一群组的消费者会负载均衡消费 Topic 的分区；auto.offset.reset 设置当消费者没有初始偏移量时的策略（earliest 从最早消息开始，latest 从最新消息开始）；enable.auto.commit 启用自动提交，auto.commit.interval.ms 设置自动提交的间隔时间。
subscribe 方法：订阅一个或多个 Topic，支持正则表达式（如 subscribe(["test_*"])）。
poll 方法：拉取消息，timeout 参数设置超时时间（毫秒），超时后返回 None。
消息处理：通过 msg.key() 和 msg.value() 获取消息的键和值，需要进行解码；msg.error() 用于判断消息是否有错误，KafkaError._PARTITION_EOF 表示到达分区末尾。

3.3.2 高级消费者（手动提交偏移量）

手动提交偏移量可以更精确地控制消息的消费进度，确保消息被成功处理后再提交偏移量，避免消息丢失。适合对数据一致性要求高的场景（如金融交易、订单处理）。

代码示例

from confluent_kafka import Consumer, KafkaError, TopicPartition
import json

# 1. 配置消费者参数（关闭自动提交）
consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "python-manual-commit-group",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False  # 关闭自动提交
}

# 2. 初始化消费者
consumer = Consumer(consumer_config)

# 3. 订阅 Topic
topic = "test_topic"
consumer.subscribe([topic])
print(f"手动提交消费者已订阅 Topic: {topic}")

# 4. 消费消息
try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue

        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print(f"分区末尾 -> {msg.topic()}-{msg.partition()}:{msg.offset()}")
            else:
                print(f"消费错误: {msg.error()}")
            continue

        # 处理消息
        key = msg.key().decode("utf-8") if msg.key() else None
        value = json.loads(msg.value().decode("utf-8"))
        print(f"消费到消息 -> Key: {key}, Value: {value}")

        # 模拟业务处理（如写入数据库、调用 API）
        # 假设这里的业务逻辑执行成功
        print("业务逻辑处理成功，准备提交偏移量")

        # 5. 手动提交偏移量
        # 方式 1：提交当前消费的消息偏移量
        consumer.commit(msg)
        print(f"偏移量提交成功 -> Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset() + 1}")

        # 方式 2：提交指定分区的偏移量（批量提交）
        # partitions = [TopicPartition(topic, msg.partition(), msg.offset() + 1)]
        # consumer.commit(partitions=partitions)

except KeyboardInterrupt:
    print("用户中断消费")
finally:
    consumer.close()
    print("消费者已关闭")

代码说明

关闭自动提交：将 enable.auto.commit 设置为 False，禁用自动提交功能。
手动提交方式：

consumer.commit(msg)：提交当前消费的消息的偏移量，Kafka 会记录该消费者群组在对应分区的偏移量为 msg.offset() + 1（下一次从该偏移量开始消费）。
consumer.commit(partitions=partitions)：批量提交多个分区的偏移量，适合批量处理消息的场景。

业务一致性：手动提交偏移量的核心优势是可以确保消息被成功处理后再提交，避免因程序崩溃导致的消息丢失。例如，在将消息写入数据库并确认写入成功后，再提交偏移量。

3.4 消费者群组与分区分配

Kafka 的消费者群组机制可以实现消息的负载均衡，当多个消费者属于同一个 group.id 时，Kafka 会将 Topic 的分区均匀分配给群组内的消费者。

示例场景
假设 test_topic 有 3 个分区，启动 2 个消费者属于同一个群组，则分区分配可能为：消费者 1 分配 2 个分区，消费者 2 分配 1 个分区。当新增一个消费者时，Kafka 会触发分区再平衡，将分区重新分配为每个消费者 1 个分区。

代码验证
启动多个上述的消费者实例（保持 group.id 相同），然后通过生产者发送消息，可以看到不同消费者消费不同分区的消息。

四、高级特性实战

4.1 事务消息

事务消息可以确保生产者发送的多条消息原子性地提交到 Kafka，同时确保消费者只消费已提交的事务消息，适合需要跨多个 Topic 或分区发送消息的场景（如分布式事务）。

代码示例

from confluent_kafka import Producer, Consumer, KafkaError, KafkaException
import json

# 生产者配置（启用事务）
producer_config = {
    "bootstrap.servers": "localhost:9092",
    "client.id": "transactional-producer",
    "acks": "all",
    "transactional.id": "test-transaction-id"  # 事务 ID，确保生产者故障恢复后的幂等性
}

# 初始化生产者并初始化事务
producer = Producer(producer_config)
producer.init_transactions()

try:
    # 开始事务
    producer.begin_transaction()

    # 发送多条消息到不同 Topic
    topic1 = "topic_tx_1"
    topic2 = "topic_tx_2"

    # 发送第一条消息
    producer.produce(topic1, value=b"Transaction Message 1", on_delivery=delivery_report)
    producer.poll(0)

    # 发送第二条消息
    producer.produce(topic2, value=b"Transaction Message 2", on_delivery=delivery_report)
    producer.poll(0)

    # 提交事务
    producer.commit_transaction()
    print("事务提交成功")

except KafkaException as e:
    print(f"事务执行失败，开始回滚: {e}")
    # 回滚事务
    producer.abort_transaction()

finally:
    producer.flush()

代码说明

事务配置：通过 transactional.id 启用事务功能，同一个 transactional.id 的生产者可以确保故障恢复后的幂等性。
事务流程：init_transactions 初始化事务，begin_transaction 开始事务，commit_transaction 提交事务，abort_transaction 回滚事务。
消费者事务隔离：消费者可以通过设置 isolation.level 参数控制是否消费未提交的事务消息，read_committed 表示只消费已提交的消息，read_uncommitted 表示消费所有消息。

4.2 SSL 加密与 SASL 认证

在生产环境中，Kafka 集群通常需要启用 SSL 加密和 SASL 认证，以确保数据传输的安全性和访问控制。

生产者配置示例（SASL/PLAIN 认证 + SSL 加密）

producer_config = {
    "bootstrap.servers": "kafka-cluster:9093",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "kafka_user",
    "sasl.password": "kafka_password",
    "ssl.ca.location": "/path/to/ca.pem",  # CA 证书路径
    "ssl.certificate.location": "/path/to/client-cert.pem",  # 客户端证书路径
    "ssl.key.location": "/path/to/client-key.pem"  # 客户端私钥路径
}

producer = Producer(producer_config)

代码说明

security.protocol 设置为 SASL_SSL，表示启用 SASL 认证和 SSL 加密。
sasl.mechanism 指定 SASL 机制（如 PLAIN、SCRAM-SHA-256）。
ssl.ca.location 指定 CA 证书路径，用于验证 Kafka 服务端证书。
ssl.certificate.location 和 ssl.key.location 指定客户端证书和私钥，用于双向认证。

五、实际业务案例：实时日志收集系统

5.1 案例背景

某电商平台需要构建一个实时日志收集系统，将用户行为日志（如浏览、点击、下单）从各个业务服务器收集到 Kafka，然后由下游的数据分析系统消费并处理这些日志。

5.2 系统架构

生产者端：业务服务器上的 Python 脚本收集用户行为日志，发送到 Kafka Topic user_behavior_topic。
Kafka 集群：存储用户行为日志，提供高吞吐量和高可用性。
消费者端：数据分析系统的 Python 脚本消费 user_behavior_topic 的日志，进行实时统计和存储。

5.3 生产者代码实现

from confluent_kafka import Producer
import json
import time
import random

# 生产者配置
producer_config = {
    "bootstrap.servers": "localhost:9092",
    "acks": "1",
    "retries": 3,
    "linger.ms": 10,
    "compression.type": "lz4"
}

producer = Producer(producer_config)

# 回调函数
def delivery_report(err, msg):
    if err:
        print(f"日志发送失败: {err}")
    else:
        print(f"日志发送成功 -> Topic: {msg.topic()}, Offset: {msg.offset()}")

# 模拟用户行为日志
def generate_user_behavior_log():
    user_ids = [f"user_{i}" for i in range(1000)]
    behaviors = ["view", "click", "add_cart", "purchase"]
    products = [f"product_{i}" for i in range(100)]

    return {
        "user_id": random.choice(user_ids),
        "behavior": random.choice(behaviors),
        "product_id": random.choice(products),
        "timestamp": int(time.time() * 1000),
        "ip": f"{random.randint(1, 255)}.{random.randint(1, 255)}.{random.randint(1, 255)}.{random.randint(1, 255)}"
    }

# 发送日志到 Kafka
topic = "user_behavior_topic"

try:
    while True:
        # 生成一条用户行为日志
        log_data = generate_user_behavior_log()
        log_value = json.dumps(log_data).encode("utf-8")

        # 发送日志
        producer.produce(topic, value=log_value, on_delivery=delivery_report)
        producer.poll(0)

        # 模拟每秒生成 10 条日志
        time.sleep(0.1)
except KeyboardInterrupt:
    producer.flush()
    print("日志生产者已停止")

5.4 消费者代码实现

from confluent_kafka import Consumer, KafkaError
import json
import pandas as pd
from collections import defaultdict

# 消费者配置
consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "user_behavior_consumer_group",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False
}

consumer = Consumer(consumer_config)
consumer.subscribe(["user_behavior_topic"])

# 统计用户行为次数
behavior_count = defaultdict(int)
# 批量处理消息的阈值
BATCH_SIZE = 100
batch_messages = []

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue

        if msg.error():
            if msg.error().code() != KafkaError._PARTITION_EOF:
                print(f"消费错误: {msg.error()}")
            continue

        # 解析日志消息
        log_data = json.loads(msg.value().decode("utf-8"))
        batch_messages.append(log_data)

        # 当批量消息达到阈值时，进行统计处理
        if len(batch_messages) >= BATCH_SIZE:
            # 转换为 DataFrame 进行分析
            df = pd.DataFrame(batch_messages)
            # 统计每种行为的次数
            behavior_stats = df["behavior"].value_counts()
            # 更新全局统计结果
            for behavior, count in behavior_stats.items():
                behavior_count[behavior] += count

            print("=" * 50)
            print("用户行为统计结果:")
            for behavior, count in behavior_count.items():
                print(f"{behavior}: {count}")
            print("=" * 50)

            # 提交偏移量
            consumer.commit(msg)
            # 清空批量消息列表
            batch_messages = []

except KeyboardInterrupt:
    print("用户中断消费")
finally:
    # 处理剩余的消息
    if batch_messages:
        df = pd.DataFrame(batch_messages)
        behavior_stats = df["behavior"].value_counts()
        for behavior, count in behavior_stats.items():
            behavior_count[behavior] += count
        print("最终统计结果:")
        for behavior, count in behavior_count.items():
            print(f"{behavior}: {count}")
    consumer.close()

5.5 案例总结

该案例利用 confluent-kafka-python 的高性能特性，实现了大规模日志的实时收集和处理。生产者端通过批量发送和压缩提高了发送效率，消费者端通过批量处理和手动提交偏移量确保了数据处理的准确性和效率。同时，该系统具有良好的扩展性，新增业务服务器只需部署生产者脚本，新增数据分析任务只需新增消费者群组。

六、相关资源链接

Pypi地址：https://pypi.org/project/confluent-kafka
Github地址：https://github.com/confluentinc/confluent-kafka-python
官方文档地址：https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html

关注我，每天分享一个实用的Python自动化工具。

实用工具

Python Kafka 开发利器：confluent-kafka-python 从入门到实战

一、confluent-kafka-python 核心概述

1.1 库的用途

1.2 工作原理

1.3 优缺点分析

1.4 License 类型

二、confluent-kafka-python 安装与环境准备

2.1 系统依赖安装

2.1.1 Linux 系统（Ubuntu/Debian）

2.1.2 Linux 系统（CentOS/RHEL）

2.1.3 macOS 系统

2.1.4 Windows 系统

2.2 Python 包安装

三、核心功能实战：生产者与消费者

3.1 Kafka 环境准备

3.2 生产者（Producer）实战

3.2.1 基础异步生产者

3.2.2 同步生产者

3.3 消费者（Consumer）实战

3.3.1 基础消费者（自动提交偏移量）

3.3.2 高级消费者（手动提交偏移量）

3.4 消费者群组与分区分配

四、高级特性实战

4.1 事务消息

4.2 SSL 加密与 SASL 认证

五、实际业务案例：实时日志收集系统

5.1 案例背景

5.2 系统架构

5.3 生产者代码实现

5.4 消费者代码实现

5.5 案例总结

六、相关资源链接

更多文章

Python实用工具Upgini：零代码特征工程与数据集增强教程

Python实用工具库excalibur：PDF表格提取与数据处理实战教程

Python实用工具：rows库快速入门与实战指南

Python实用工具：Camelot库——轻松提取PDF表格数据的完整指南