import time

from kafka import KafkaProducer
import json
import random



def producer_message(topic,key,data_list):
    # Kafka 配置
    bootstrap_servers = ['172.16.3.6:9092']  # 替换为你的Kafka地址
    # topic = 'aidso_monitor'  # 目标主题

    # 创建生产者，优化批量发送参数
    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        key_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode('utf-8'),
        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode('utf-8'),
        # 批量发送优化参数
        batch_size=16384,
        linger_ms=5,
        compression_type='snappy'
    )

    try:
        for i in data_list:
            producer.send(topic, key=random.randint(0, key), value=i)
        producer.flush()
    except Exception as e:
        print(f"发送失败：{str(e)}")

if __name__ == '__main__':
    ...
