from aiokafka import AIOKafkaProducer

# Kafka 配置
KAFKA_BOOTSTRAP_SERVERS = '127.0.0.1:19092'
# KAFKA_TOPIC = 'binance_agg_trade'

class KafkaProducerSingleton:
    _instance = None
    _producer = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(KafkaProducerSingleton, cls).__new__(cls)
        return cls._instance

    async def get_producer(self):
        if self._producer is None:
            self._producer = AIOKafkaProducer(
                    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
                    key_serializer=lambda k: k.encode('utf-8'),
                    value_serializer=lambda v: v.encode('utf-8'),
                    linger_ms=100
                    )
            await self._producer.start()
        return self._producer

    async def stop_producer(self):
        if self._producer is not None:
            await self._producer.stop()
            self._producer = None

# 初始化单例实例
_kafka_producer_singleton = KafkaProducerSingleton()

async def send_to_kafka(topic, data, key=None):
    """将数据发送到 Kafka 主题"""
    producer = await _kafka_producer_singleton.get_producer()
    #print(topic, data, key)
    try:
        await producer.send(topic, data,key=key)
    except Exception as e:
        print("error: ", topic, data, key,e)