def delivery_callback(err, msg):
    if err:
        print(f'消息发送失败: {err}')
    else:

print(f'消息已发送到 {msg.topic()} [分区 {msg.partition()}]')

# 批量消息生产
for i in range(10000):
    data = {'id': i, 'value': f'消息-{i}'}
    producer.produce(

'output-topic',
        key=str(i).encode(),
        value=json.dumps(data).encode(),
        callback=delivery_callback
    )

    # 每1000条消息触发一次回调
    if i % 1000 == 0:
        producer.poll(0)

# 等待所有消息被发送
producer.flush(30)

该模式利用了Kafka生产者的异步特性。通过定期调用poll()并使用回调，我们可以在保持高吞吐量的同时，确保对交付状态的可见性。 针对不同用例的消费策略

# 简单消费循环
try:
    while True:

msg = consumer.poll(1.0)
if msg is None:
    continue

if msg.error():
    print(f'消费者错误: {msg.error()}')
    continue

# 处理消息
    data = json.loads(msg.value().decode('utf-8'))

print(f'处理后的消息: {data})

# 手动提交
        consumer.commit(msg)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

对于流中的批处理，我使用以下方法： batch = []

batch_size = 100
last_commit = time.time()

while True:
    msg = consumer.poll(0.1)

    if msg and not msg.error():
        batch.append(json.loads(msg.value().decode('utf-8')))

# 当批处理满或超过时间阈值时处理批次
    current_time = time.time()
    if len(batch) >= batch_size or (batch and current_time - last_commit > 5):
        process_batch(batch)
        batch = []
        consumer.commit()
        last_commit = current_time

from confluent_kafka import Producer, Consumer, KafkaException, TIMESTAMP_CREATE_TIME
import uuid

# 事务生产者
producer_conf = {
    'bootstrap.servers': 'kafka:9092',

'transactional.id': f'my-transactional-producer-{uuid.uuid4()}' } producer = Producer(producer_conf) producer.init_transactions() try: # 开始事务  producer.begin_transaction()

# 在同一事务中处理和生产
for i in range(100):

producer.produce(‘output-topic‘, key=str(i).encode(), value=f‘value-{i}‘.encode())

# 提交事务
producer.commit_transaction()
except KafkaException as e:

# 出错时中止
producer.abort_transaction()
raise

该模式确保事务中的所有消息要么全部写入，要么全部不写入，从而防止导致不一致状态的部分更新。

import asyncio
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json

python
async def produce_messages():
producer = AIOKafkaProducer(
bootstrap_servers=‘kafka:9092‘,
value_serializer=lambda v: json.dumps(v).encode()
)
await producer.start()

try:
for i in range(1000):
在这段代码中，我们定义了一个异步函数 `produce_messages`，它创建了一个 `AIOKafkaProducer` 实例，并指定了 Kafka 服务器的地址和消息值的序列化方式。接着，我们调用 `start` 方法来启动生产者。在 `try` 块中，我们使用一个循环来生成 1000 条消息。

await producer.send('async-topic', {'number': i})
            if i % 100 == 0:
                await asyncio.sleep(0.1)  # 模拟其他工作
    finally:
        await producer.stop()

async def consume_messages():
    consumer = AIOKafkaConsumer(
        'async-topic',

bootstrap_servers='kafka:9092',
value_deserializer=lambda m: json.loads(m.decode())
)
await consumer.start()

try:
    async for msg in consumer:
        # 异步处理消息
        await process_message(msg.value)
finally:
    await consumer.stop()

async def process_message(value):
    # 模拟异步处理
    await asyncio.sleep(0.01)
    print(f"处理完成: {value}")

# 同时运行生产者和消费者
async def main():
    await asyncio.gather(
        produce_messages(),
        consume_messages()
    )

asyncio.run(main())

import faust

app = faust.App(
    'my-stream-app',
    broker='kafka://kafka:9092',

value_serializer='json',
)

# 定义一个用于类型检查的模型
class Order(faust.Record):
    customer_id: str
    product_id: str
    price: float
    quantity: int

# 定义输入和输出主题
orders_topic = app.topic('orders', value_type=Order)
revenue_topic = app.topic('revenue')

# 创建一个用于存储状态的表
customer_revenue = app.Table('customer_revenue', default=float)

# 定义流处理器
@app.agent(orders_topic)
async def process_order(orders):
    async for order in orders:
        # 计算订单价值
        order_value = order.price * order.quantity

        # 更新客户收入

        customer_revenue[order.customer_id] += order_value

        # 将结果转发到输出主题
        await revenue_topic.send(
            key=order.customer_id,
            value={
                'customer_id': order.customer_id,
                'revenue': customer_revenue[order.customer_id],
                'last_order_value': order_value
            }
        )

if __name__ == '__main__':
    app.main()

import faust
from datetime import timedelta

app = faust.App(
    'windowing-example',
    broker='kafka://kafka:9092',
)

class PageView(faust.Record):
    user_id: str
    page_id: str
    timestamp: float

page_views_topic = app.topic(‘page-views‘, value_type=PageView)

# 创建一个具有1分钟滚动窗口的表
page_view_counts = app.Table(
‘page-view-counts‘,
default=int,
).tumbling(timedelta(minutes=1))

@app.agent(page_views_topic)
async def count_page_views(views):

async for view in views:
        # 在当前窗口中为此页面增加计数
        page_view_counts[view.page_id] += 1

        # 当前窗口的计数
        current_count = page_view_counts[view.page_id].current()
        print(f"页面 {view.page_id} 在当前窗口中有 {current_count} 次浏览")

import faust

app = faust.App('stream-join-example', broker='kafka://kafka:9092')

from confluent_kafka import Producer, Consumer, KafkaError
import json

# 配置死信队列
def setup_dlq_producer():
    return Producer({
        'bootstrap.servers': 'kafka:9092',
        'client.id': 'error-handler'

import time
import prometheus_client
from confluent_kafka.admin import AdminClient, ClusterMetadata
from prometheus_client import start_http_server, Counter, Gauge

# 设置Prometheus指标

messages_processed = Counter('kafka_messages_processed_total', 
                            '总处理消息数', 
                            ['topic', 'result'])
processing_time = Gauge('kafka_message_processing_seconds', 
                       '处理消息所花费的时间')
consumer_lag = Gauge('kafka_consumer_lag', 
                    '消费者消息滞后',

['topic', 'partition'])

# 暴露指标
start_http_server(8000)

# 用于延迟监控的管理客户端
admin_client = AdminClient({'bootstrap.servers': 'kafka:9092'})

def get_consumer_lag(consumer, topic_partitions):
    # 获取主题分区的结束偏移量
    watermarks = {}
    for tp in topic_partitions:

low, high = consumer.get_watermark_offsets(tp)
        watermarks[tp] = high

    # 获取已提交的偏移量
    committed = consumer.committed(topic_partitions)

    # 计算并报告延迟
    for tp, offset in committed.items():
        if offset and tp in watermarks:

lag = watermarks[tp] – offset.offset
consumer_lag.labels(
topic=tp.topic,
partition=tp.partition
).set(lag)

# 在处理循环中使用
def monitored_processing():
with processing_time.time():
try:
# 处理消息
result = “success“

except Exception:
            result = "错误"
            raise
        finally:
            messages_processed.labels(topic="我的主题", result=result).inc()

这些指标提供了对您应用程序性能和健康状况的可视化。

部署模式

我发现以下部署方法对 Python Kafka 应用程序特别有效：

# 带有优雅关闭的工作模式
import signal
import sys

class KafkaWorker:
    def __init__(self):
        self.consumer = Consumer({
            'bootstrap.servers': 'kafka:9092',
            'group.id': 'worker-group',