python-kafka-module-skill

Python Kafka Module Skill

面向已有 FastAPI 项目的开发者，快速集成 Kafka 能力。

能力清单

能力	说明
生产者	同步/异步发送消息、消息分区、消息key
消费者	消费监听、消息重试、消费者组
消息序列化	JSON
事务消息	Kafka 事务
错误处理	消息发送/消费错误处理

触发场景

用户说"帮我加 Kafka"或"集成 Kafka"时触发。

依赖配置

pip install aiokafka

默认方法封装

1. 生产者

# kafka_producer.py
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
from typing import Optional, Callable, Any
import json
import logging

logger = logging.getLogger(__name__)

class KafkaProducer:
    def __init__(self, bootstrap_servers: str = "localhost:9092"):
        self.bootstrap_servers = bootstrap_servers
        self._producer: Optional[AIOKafkaProducer] = None
    
    async def start(self):
        """启动生产者"""
        self._producer = AIOKafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None
        )
        await self._producer.start()
        logger.info("Kafka 生产者已启动")
    
    async def stop(self):
        """停止生产者"""
        if self._producer:
            await self._producer.stop()
            logger.info("Kafka 生产者已停止")
    
    async def send(self, topic: str, value: Any, key: Optional[str] = None) -> str:
        """发送消息（同步）"""
        if not self._producer:
            raise RuntimeError("生产者未启动")
        
        future = await self._producer.send_and_wait(topic, value, key=key)
        return f"{future.topic}-{future.partition}-{future.offset}"
    
    async def send_async(self, topic: str, value: Any, key: Optional[str] = None, callback: Optional[Callable] = None):
        """发送消息（异步）"""
        if not self._producer:
            raise RuntimeError("生产者未启动")
        
        await self._producer.send(topic, value, key=key)
        if callback:
            # 注册回调
            pass
    
    async def send_json(self, topic: str, data: dict, key: Optional[str] = None):
        """发送 JSON 消息"""
        return await self.send(topic, data, key)
    
    async def send_messages(self, topic: str, messages: list):
        """批量发送消息"""
        if not self._producer:
            raise RuntimeError("生产者未启动")
        
        for msg in messages:
            await self._producer.send(topic, msg)

# 全局实例
producer = KafkaProducer()