# -*- encoding:utf-8 -*-

'''
@Author : dingjiawen
@Date : 2023/5/16 20:15
@Usage : 获取kakfa的producer和Consumer，以及发布信息和订阅信息
@Desc :  创建kafka的工具类
'''

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json

KAFKA_SERVER = ["Ding202:9092", "Ding203:9092", "Ding204:9092"]


def getKafkaProducer(key_serializer=None, value_serializer=None):
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    # producer = KafkaProducer(
    #     bootstrap_servers=['192.168.118.202:9092'],
    #     key_serializer=lambda v: json.dumps(v).encode('utf-8'),
    #     value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    # )

    producer = KafkaProducer(
        bootstrap_servers=KAFKA_SERVER,
        key_serializer=key_serializer,
        value_serializer=value_serializer
    )

    return producer

def getKafkaProducerTest( key_serializer=None, value_serializer=None):
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    # producer = KafkaProducer(
    #     bootstrap_servers=['192.168.118.202:9092'],
    #     key_serializer=lambda v: json.dumps(v).encode('utf-8'),
    #     value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    # )

    producer = KafkaProducer(
        bootstrap_servers=KAFKA_SERVER,
        key_serializer=key_serializer,
        value_serializer=value_serializer
    )

    return producer

def getKafkaConsumer(topic,group_id ,key_serializer=None, value_serializer=None):
    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
    consumer = KafkaConsumer(
        topic,
        group_id=group_id,
        bootstrap_servers=KAFKA_SERVER,
        auto_offset_reset='latest'
    )

    return consumer