conf.ini文件配置主题信息

[topic_name]
topic = topic_name
bootstrap_servers = X.X.X.X:9092,X.X.X.X:9092

获取配置

import configparser, os
class Base_confi():
    """默认读取conf_data下的conf.ini文件,初始化读取conf.ini配置文件"""
    def __init__(self):
        self.configfile_path = os.path.join(os.path.dirname(__file__), '..', 'config/conf.ini')
        self.conf = configparser.ConfigParser()
        self.conf.read(self.configfile_path)

    """获取指定节点下指定option的值，返回为字符串类型"""
    def config_get(self, title, value):
        return self.conf.get(title, value)

    """返回conf.ini文件路径"""
    def config_path(self):
        return self.configfile_path

    """修改conf.ini文件数据，如果section不存在的话，先进行判断然后进行添加再修改"""
    def config_modify(self, section, option, value):
        if not self.conf.has_section(section):
            self.conf.add_section(section)
        self.conf.set(section, option, value)
        with open(self.configfile_path, "w") as f:
            self.conf.write(f)

发送消息方法

class KafkaProduct():
    def __init__(self, kafka_section):
        self.data = Base_confi()
        self.topic = self.data.config_get(kafka_section, "topic")
        self.bootstrap_servers = self.data.config_get(kafka_section, "bootstrap_servers")
        self.producer = KafkaProducer(bootstrap_servers=self.bootstrap_servers)

    def send_message(self, kafka_data):
        parmas_message = json.dumps(kafka_data)
        producer = self.producer
        try:
            producer.send(self.topic, parmas_message.encode('utf-8'))
            producer.flush()
            return {"success": True}
        except KafkaError as e:
            print("kafka error is {}".format(e))
            return {"success":False,"message":format(e)}

发送消息

def bag():
    kafkadata = {
        "lastTime": "2024-03-15 10:58:02",
        "endTime":  "2024-03-15 10:58:02"
    }
    product=KafkaProduct("topic_name")
    res =product.send_message(kafkadata)
    print("kafka返回：" + str(res))
    return res