from kafka import KafkaProducer, KafkaConsumer
import json
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ------------------- 生产者 -------------------
class SimpleKafkaProducer:
def __init__(self, bootstrap_servers=["localhost:9092"]):
self.producer = KafkaProducer(
bootstrap_servers=bootstrap_servers,
value_serializer=lambda v: json.dumps(v).encode("utf-8"), # 自动序列化JSON
acks="1", # Leader确认即可，平衡性能和可靠性
retries=3, # 失败重试3次
linger_ms=5, # 延迟5ms凑批次，提高吞吐量
)
def send(self, topic: str, value: dict, key: str = None):
"""发送单条消息，key可以控制消息发到固定分区（比如同一用户的事件发到同一区）"""
try:
future = self.producer.send(
topic,
value=value,
key=key.encode("utf-8") if key else None
)
# 等待发送完成（可选，生产环境可以异步）
record_metadata = future.get(timeout=10)
logger.info(
f"消息发送成功：Topic={record_metadata.topic}，分区={record_metadata.partition}，偏移量={record_metadata.offset}"
)
except Exception as e:
logger.error(f"消息发送失败：{e}")
raise
def close(self):
self.producer.flush() # 刷新缓冲区
self.producer.close()
# ------------------- 消费者 -------------------
class SimpleKafkaConsumer:
def __init__(self, topics: list, group_id: str, bootstrap_servers=["localhost:9092"]):
self.consumer = KafkaConsumer(
*topics,
bootstrap_servers=bootstrap_servers,
group_id=group_id,
value_deserializer=lambda m: json.loads(m.decode("utf-8")), # 自动反序列化JSON
auto_offset_reset="earliest", # 新组从最早的消息读
enable_auto_commit=True, # 自动提交偏移量（生产环境建议手动）
auto_commit_interval_ms=1000,
)
def consume(self, callback):
"""循环消费，传入回调函数处理消息"""
logger.info(f"开始消费：Topics={self.consumer.subscription()}，Group={self.consumer.config['group_id']}")
try:
for msg in self.consumer:
callback(msg.value)
except KeyboardInterrupt:
logger.info("消费已停止")
finally:
self.consumer.close()
# ------------------- 使用示例 -------------------
if __name__ == "__main__":
# 1. 发送订单消息
producer = SimpleKafkaProducer()
producer.send(
topic="order-events",
value={"order_id": 1001, "user_id": 123, "product_id": 456, "amount": 99.9},
key=str(123), # 同一用户的订单发到同一分区
)
producer.close()
# 2. 消费订单消息（单独开一个终端运行）
# def process_order(msg):
# logger.info(f"收到订单：{msg}")
# consumer = SimpleKafkaConsumer(topics=["order-events"], group_id="order-consumer-1")
# consumer.consume(process_order)
