python之kafka消費

使用python3第三方工具，實現kafka消費

# -*- coding: utf-8 -*-

import uuid
import json
from kafka import KafkaConsumer
from xxxxxx import MessageToDict
from xxx import ObjectInfo

import sys
import codecs

sys.stdout = codecs.getwriter("utf-8")(sys.stdout.detach())


class ReadKafkaContent(object):
    @staticmethod
    def deserialize(msg):
        """
        反序列化
        :param msg:
        :return:
        """
        pb_obj = ObjectInfo()
        pb_obj.Clear()
        pb_obj.ParseFromString(msg.value)
        return MessageToDict(pb_obj, including_default_value_fields=True, preserving_proto_field_name=True)

    def consume_msg(self, consumer_obj):
        """
        逐條消費，返回反序列化后的內容
        :param consumer_obj:
        :return:
        """
        try:
            while True:
                msg = next(consumer_obj, None)
                if not msg:
                    continue
                content = self.deserialize(msg)
                return content
        except Exception as ex:
            print(u"消費kafka錯誤,退出測試")
            return None

    def entry(self, topic, ip, count=10, log="log_read_kafka_content.json"):
        """

        :param topic:topic
        :param ip:ip
        :param count:查詢kafka數據數量,默認10條
        :param log:內容保存地址,默認
        :return:
        """
        print(u"開始......")
        try:
            # 創建kafka消費對象
            print(u"創建kafka消費對象...")
            consumer = KafkaConsumer(topic, group_id=uuid.uuid4().hex,
                                     bootstrap_servers=[ip],
                                     auto_offset_reset="latest", consumer_timeout_ms=3 * 1000)
        except Exception as ex:
            print(u"連接kafka失敗!")
            return False
        print(u"kafka消費對象創建成功.")

        with open(log, "w") as f:
            for i in range(count):
                print(u"開始消費第%s條數據..." % str(i + 1))
                content = self.consume_msg(consumer)
                if not content:
                    return False

                # dict轉json保存數據內容
                content_json = json.dumps(content, ensure_ascii=False, indent=4)
                f.write(content_json)
                f.write("\n\n")
                print(u"第%s條數據寫入完成." % str(i + 1))

        print(u"完成.")