python消费kafka收到的json消息

在Python中，要消费Kafka接收到的JSON消息，首先需要安装一些库，如`confluent-kafka-python`，它是Apache Kafka官方推荐的Python客户端。下面是基本步骤：

1. **安装依赖**:

   pip install confluent-kafka

2. **创建消费者**:
使用`ConfluentKafkaConsumer`类创建一个消费者实例，并指定主题、组名等信息：

   from confluent_kafka import Consumer

   consumer = Consumer({
       'bootstrap.servers': 'your.kafka.server:9092',  # 替换为实际Kafka服务器地址
       'group.id': 'my_group',  # 消费者组ID
       'auto.offset.reset': 'earliest' or 'latest',  # 设置偏移量恢复策略
       'value_deserializer': lambda m: json.loads(m.decode('utf-8'))  # 解析消息内容为字典
   })

3. **订阅主题**:

   consumer.subscribe(['your_topic'])  # 替换为主题名称

4. **开始消费**:

   while True:
       message = consumer.poll(1.0)  # 等待接收消息，单位秒
       if message is None:
           continue
       elif message.error():
           print(f"Error: {message.error()}")
       else:
           data = message.value()
           process_json_data(data)  # 自定义处理函数，处理接收到的JSON数据

5. **处理函数**:
这里`process_json_data`是你自定义的函数，用于解析并操作接收到的JSON数据。