telegram_crawler/utils/push_kafka.py

from kafka import KafkaProducer, KafkaConsumer
from config import logger


class SKafka():
    def __init__(self, bootstrap_servers: str, api_version=None, is_json_schema: bool = False,encoding: str = "utf-8",
                 security_protocol: str = "PLAINTEXT", sasl_mechanism: str = None,
                 sasl_plain_username: str = None, sasl_plain_password: str = None):
        self.is_json_schema = is_json_schema

        self.encoding = encoding

        self.producer = KafkaProducer(bootstrap_servers=bootstrap_servers,api_version=api_version,
                                      security_protocol=security_protocol,sasl_mechanism=sasl_mechanism,
                                      sasl_plain_username=sasl_plain_username,sasl_plain_password=sasl_plain_password)

    def sync_producer(self, topic, message):
        count = 0
        for date in message:
            if date:
                date = date.encode()
                self.producer.send(topic, date)
                count += 1
        logger.info('push {} success,date length:{} vail date length: {} '.format(topic, len(message), count))
            # self.producer.flush()

    def close_producer(self):
        try:
            self.producer.close()
            print("connect close")
        except Exception as e:
            print(e)
        pass






# kafka设置了密码，没有就不用设置那么多
if __name__ == '__main__':

    # kp = SKafka(bootstrap_servers='172.18.1.119:9992')
    hosts = "172.18.1.119:9992"  # 连接hosts
    topic = "dazhongdianping"
    # producer = KafkaProducer(bootstrap_servers='172.18.1.119:9992')
    consumer = KafkaConsumer('dazhongdianping', bootstrap_servers='172.18.1.119:9992')
    for message in consumer:
        # 解析并处理消息
        print(f"Received message: {message.value.decode('utf-8')}")