自学教程：python操作kafka的详细步骤

pip install confluent-kafka==1.9.2

kafka_setting = {    'sasl_plain_username': 'XXX',   #如果是默认接入点实例，请删除该配置。    'sasl_plain_password': 'XXX',   #如果是默认接入点实例，请删除该配置。    'bootstrap_servers': '[xxx,xxx,xxx]',    'topic_name': 'XXX',    'group_name': 'XXX'}

# -*- coding: utf-8 -*- import jsonimport jsonimport msgpackfrom loguru import loggerfrom kafka import KafkaProducerfrom kafka.errors import KafkaError def kfk_produce_1():    """        发送 json 格式数据    :return:    """    producer = KafkaProducer(        bootstrap_servers='ip:9092',        value_serializer=lambda v: json.dumps(v).encode('utf-8')    )    producer.send('test_topic', {'key1': 'value1'})  def kfk_produce_2():    """        发送 string 格式数据    :return:    """    producer = KafkaProducer(bootstrap_servers='xxxx:x')    data_dict = {        "name": 'king',        'age': 100,        "msg": "Hello World"    }    msg = json.dumps(data_dict)    producer.send('test_topic', msg, partition=0)    producer.close()  def kfk_produce_3():    producer = KafkaProducer(bootstrap_servers=['broker1:1234'])     # Asynchronous by default ( 默认是异步发送 )    future = producer.send('my-topic', b'raw_bytes')     # Block for 'synchronous' sends    try:        record_metadata = future.get(timeout=10)    except KafkaError:        # Decide what to do if produce request failed...        logger.error(KafkaError)        pass     # Successful result returns assigned partition and offset    print(record_metadata.topic)    print(record_metadata.partition)    print(record_metadata.offset)     # produce keyed messages to enable hashed partitioning    producer.send('my-topic', key=b'foo', value=b'bar')     # encode objects via msgpack    producer = KafkaProducer(value_serializer=msgpack.dumps)    producer.send('msgpack-topic', {'key': 'value'})     # produce json messages    producer = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))    producer.send('json-topic', {'key': 'value'})     # produce asynchronously    for _ in range(100):        producer.send('my-topic', b'msg')     def on_send_success(record_metadata=None):        print(record_metadata.topic)        print(record_metadata.partition)        print(record_metadata.offset)     def on_send_error(excp=None):        logger.error('I am an errback', exc_info=excp)        # handle exception     # produce asynchronously with callbacks    producer.send('my-topic', b'raw_bytes').add_callback(on_send_success).add_errback(on_send_error)     # block until all async messages are sent    producer.flush()     # configure multiple retries    producer = KafkaProducer(retries=5) if __name__ == '__main__':    kfk_produce_1()    kfk_produce_2()    pass

# -*- coding: utf-8 -*- import jsonimport msgpackfrom kafka import KafkaConsumer # To consume latest messages and auto-commit offsetsconsumer = KafkaConsumer(    'my-topic', group_id='my-group',    bootstrap_servers=['localhost:9092'])for message in consumer:    # message value and key are raw bytes -- decode if necessary!    # e.g., for unicode: `message.value.decode('utf-8')`    info = f'{message.topic}:{message.partition}:{message.offset}: key={message.key}, value={message.value}'    print(info) # consume earliest available messages, don't commit offsetsKafkaConsumer(auto_offset_reset='earliest', enable_auto_commit=False) # consume json messagesKafkaConsumer(value_deserializer=lambda m: json.loads(m.decode('ascii'))) # consume msgpackKafkaConsumer(value_deserializer=msgpack.unpackb) # StopIteration if no message after 1sec ( 没有消息时，1s后停止消费 )KafkaConsumer(consumer_timeout_ms=1000) # Subscribe to a regex topic patternconsumer = KafkaConsumer()consumer.subscribe(pattern='^awesome.*') # Use multiple consumers in parallel w/ 0.9 kafka brokers# typically you would run each on a different server / process / CPUconsumer1 = KafkaConsumer(    'my-topic', group_id='my-group',    bootstrap_servers='my.server.com')consumer2 = KafkaConsumer(    'my-topic', group_id='my-group',    bootstrap_servers='my.server.com')

# -*- coding: utf-8 -*- import timeimport jsonimport ujsonimport randomfrom loguru import loggerfrom kafka import KafkaProducer, KafkaConsumer  class KafkaOperate(object):     def __init__(self, bootstrap_servers=None):        if not bootstrap_servers:            raise Exception('bootstrap_servers is None')         self.__bootstrap_servers = None        if isinstance(bootstrap_servers, str):            ip_port_string = bootstrap_servers.strip()            if ',' in ip_port_string:                self.__bootstrap_servers = ip_port_string.replace(' ', '').split(',')            else:                self.__bootstrap_servers = [ip_port_string]         self.kafka_producer = None        self.kafka_consumer = None         pass     def __del__(self):        pass     def kfk_consume(self, topic_name=None, group_id='my_group'):        if not self.kafka_consumer:            self.kafka_consumer = KafkaConsumer(                topic_name, group_id=group_id,                bootstrap_servers=self.__bootstrap_servers,                auto_offset_reset='earliest',            )        count = 0        for msg in self.kafka_consumer:            count += 1            # message value and key are raw bytes -- decode if necessary!            # e.g., for unicode: `message.value.decode('utf-8')`            info = f'[{count}] {msg.topic}:{msg.partition}:{msg.offset}: key={msg.key}, value={msg.value.decode("utf-8")}'            logger.info(info)            time.sleep(1)     def __kfk_produce(self, topic_name=None, data_dict=None, partition=None):        """            如果想要多线程进行消费，可以设置 发往不通的 partition            有多少个 partition 就可以启多少个线程同时进行消费，        :param topic_name:        :param data_dict:        :param partition:        :return:        """        if not self.kafka_producer:            self.kafka_producer = KafkaProducer(                bootstrap_servers=self.__bootstrap_servers,                client_id='my_group',                value_serializer=lambda v: json.dumps(v).encode('utf-8')            )        # data_dict = {        #     "name": 'king',        #     'age': 100,        #     "msg": "Hello World"        # }        if partition:            self.kafka_producer.send(                topic=topic_name,                 value=data_dict,                # key='count_num',  # 同一个key值，会被送至同一个分区                partition=partition            )        else:            self.kafka_producer.send(topic_name, data_dict)        pass     def kfk_produce_one(self, topic_name=None, data_dict=None, partition=None, partition_count=1):        partition = partition if partition else random.randint(0, partition_count-1)        self.__kfk_produce(topic_name=topic_name, data_dict=data_dict, partition=partition)        self.kafka_producer.flush()     def kfk_produce_many(self, topic_name=None, data_dict_list=None, partition=None, partition_count=1, per_count=100):        count = 0        for data_dict in data_dict_list:            partition = partition if partition else count % partition_count            self.__kfk_produce(topic_name=topic_name, data_dict=data_dict, partition=partition)            if 0 == count % per_count:                self.kafka_producer.flush()            count += 1        self.kafka_producer.flush()        pass     @staticmethod    def get_consumer(group_id: str, bootstrap_servers: list, topic: str, enable_auto_commit=True) -> KafkaConsumer:        topics = tuple([x.strip() for x in topic.split(',') if x.strip()])        if enable_auto_commit:            return KafkaConsumer(                *topics,                group_id=group_id,                bootstrap_servers=bootstrap_servers,                auto_offset_reset='earliest',                # fetch_max_bytes=FETCH_MAX_BYTES,                # connections_max_idle_ms=CONNECTIONS_MAX_IDLE_MS,                # max_poll_interval_ms=KAFKA_MAX_POLL_INTERVAL_MS,                # session_timeout_ms=SESSION_TIMEOUT_MS,                # max_poll_records=KAFKA_MAX_POLL_RECORDS,                # request_timeout_ms=REQUEST_TIMEOUT_MS,                # auto_commit_interval_ms=AUTO_COMMIT_INTERVAL_MS,                value_deserializer=lambda m: ujson.loads(m.decode('utf-8'))            )        else:            return KafkaConsumer(                *topics,                group_id=group_id,                bootstrap_servers=bootstrap_servers,                auto_offset_reset='earliest',                # fetch_max_bytes=FETCH_MAX_BYTES,                # connections_max_idle_ms=CONNECTIONS_MAX_IDLE_MS,                # max_poll_interval_ms=KAFKA_MAX_POLL_INTERVAL_MS,                # session_timeout_ms=SESSION_TIMEOUT_MS,                # max_poll_records=KAFKA_MAX_POLL_RECORDS,                # request_timeout_ms=REQUEST_TIMEOUT_MS,                enable_auto_commit=enable_auto_commit,                value_deserializer=lambda m: ujson.loads(m.decode('utf-8'))            )     @staticmethod    def get_producer(bootstrap_servers: list):        return KafkaProducer(bootstrap_servers=bootstrap_servers, retries=5)  if __name__ == '__main__':    bs = '10.10.10.10:9092'    kafka_op = KafkaOperate(bootstrap_servers=bs)    kafka_op.kfk_consume(topic_name='001_test')    pass

# -*- coding:utf-8 -*- import jsonfrom kafka import KafkaConsumer, KafkaProducer  class KProducer:    def __init__(self, bootstrap_servers, topic):        """        kafka 生产者        :param bootstrap_servers: 地址        :param topic:  topic        """        self.producer = KafkaProducer(            bootstrap_servers=bootstrap_servers,            value_serializer=lambda m: json.dumps(m).encode('ascii'), )  # json 格式化发送的内容        self.topic = topic     def sync_producer(self, data_li: list):        """        同步发送 数据        :param data_li:  发送数据        :return:        """        for data in data_li:            future = self.producer.send(self.topic, data)            record_metadata = future.get(timeout=10)  # 同步确认消费            partition = record_metadata.partition  # 数据所在的分区            offset = record_metadata.offset  # 数据所在分区的位置            print('save success, partition: {}, offset: {}'.format(partition, offset))     def asyn_producer(self,  data_li: list):        """        异步发送数据        :param data_li:发送数据        :return:        """        for data in data_li:            self.producer.send(self.topic, data)        self.producer.flush()  # 批量提交     def asyn_producer_callback(self,  data_li: list):        """        异步发送数据 + 发送状态处理        :param data_li:发送数据        :return:        """        for data in data_li:            self.producer.send(self.topic, data).add_callback(self.send_success).add_errback(self.send_error)        self.producer.flush()  # 批量提交     def send_success(self, *args, **kwargs):        """异步发送成功回调函数"""        print('save success')        return     def send_error(self, *args, **kwargs):        """异步发送错误回调函数"""        print('save error')        return     def close_producer(self):        try:            self.producer.close()        except:            pass if __name__ == '__main__':     send_data_li = [{"test": 1}, {"test": 2}]    kp = KProducer(topic='topic', bootstrap_servers='127.0.0.1:9001,127.0.0.1:9002')     # 同步发送    kp.sync_producer(send_data_li)     # 异步发送    # kp.asyn_producer(send_data_li)     # 异步+回调    # kp.asyn_producer_callback(send_data_li)        kp.close_producer()

from kafka import KafkaConsumer consumer = KafkaConsumer('test_rhj', bootstrap_servers=['xxxx:x'])for msg in consumer:    recv = "%s:%d:%d: key=%s value=%s" % (        msg.topic, msg.partition, msg.offset, msg.key, msg.value    )    print(recv)

from kafka import KafkaConsumer consumer = KafkaConsumer(    'test_rhj',     group_id='123456',     bootstrap_servers=['10.43.35.25:4531'])for msg in consumer:    recv = "%s:%d:%d: key=%s value=%s" % (        msg.topic, msg.partition, msg.offset, msg.key, msg.value    )    print(recv)

from kafka import KafkaConsumerfrom kafka.structs import TopicPartition consumer = KafkaConsumer(    group_id='123456', bootstrap_servers=['10.43.35.25:4531'])consumer.assign(    [        TopicPartition(topic='test_rhj', partition=0),        TopicPartition(topic='test_rhj', partition=1)    ]) print(consumer.partitions_for_topic("test_rhj"))  # 获取test主题的分区信息print(consumer.assignment())print(consumer.beginning_offsets(consumer.assignment())) consumer.seek(TopicPartition(topic='test_rhj', partition=0), 0)for msg in consumer:    recv = "%s:%d:%d: key=%s value=%s" % (        msg.topic, msg.partition, msg.offset, msg.key, msg.value    )    print(recv)

from kafka import KafkaConsumerimport time consumer = KafkaConsumer(group_id='123456', bootstrap_servers=['10.43.35.25:4531'])consumer.subscribe(topics=('test_rhj',))index = 0while True:    msg = consumer.poll(timeout_ms=5)  # 从kafka获取消息    print(msg)    time.sleep(2)    index += 1    print('--------poll index is %s----------' % index)

# coding:utf8from kafka import KafkaConsumer # 创建一个消费者，指定了topic,group_id,bootstrap_servers# group_id: 多个拥有相同group_id的消费者被判定为一组，#            一条数据记录只会被同一个组中的一个消费者消费# bootstrap_servers：kafka的节点，多个节点使用逗号分隔# 这种方式只会获取新产生的数据 bootstrap_server_list = [    '192.168.70.221:19092',    '192.168.70.222:19092',    '192.168.70.223:19092'] consumer = KafkaConsumer(    # kafka 集群地址    bootstrap_servers=','.join(bootstrap_server_list),    group_id="my.group",  # 消费组id    enable_auto_commit=True,  # 每过一段时间自动提交所有已消费的消息（在迭代时提交）    auto_commit_interval_ms=5000,  # 自动提交的周期（毫秒）) consumer.subscribe(["my.topic"])  # 消息的主题，可以指定多个 for msg in consumer:  # 迭代器，等待下一条消息    print(msg)  # 打印消息

# coding:utf-8 import osimport sysimport threadingfrom kafka import KafkaConsumer, TopicPartition, OffsetAndMetadatafrom collections import OrderedDict threads = [] class MyThread(threading.Thread):    def __init__(self, thread_name, topic, partition):        threading.Thread.__init__(self)        self.thread_name = thread_name        self.partition = partition        self.topic = topic     def run(self):        print("Starting " + self.name)        consumer(self.thread_name, self.topic, self.partition)     def stop(self):        sys.exit()  def consumer(thread_name, topic, partition):    broker_list = 'ip1:9092,ip2:9092'     '''    fetch_min_bytes（int） - 服务器为获取请求而返回的最小数据量，否则请等待    fetch_max_wait_ms（int） - 如果没有足够的数据立即满足fetch_min_bytes给出的要求，服务器在回应提取请求之前将阻塞的最大时间量（以毫秒为单位）    fetch_max_bytes（int） - 服务器应为获取请求返回的最大数据量。这不是绝对最大值，如果获取的第一个非空分区中的第一条消息大于此值，                则仍将返回消息以确保消费者可以取得进展。注意：使用者并行执行对多个代理的提取，因此内存使用将取决于包含该主题分区的代理的数量。                支持的Kafka版本> = 0.10.1.0。默认值：52428800（50 MB）。    enable_auto_commit（bool） - 如果为True，则消费者的偏移量将在后台定期提交。默认值：True。    max_poll_records（int） - 单次调用中返回的最大记录数poll()。默认值：500    max_poll_interval_ms（int） - poll()使用使用者组管理时的调用之间的最大延迟 。这为消费者在获取更多记录之前可以闲置的时间量设置了上限。                  如果 poll()在此超时到期之前未调用，则认为使用者失败，并且该组将重新平衡以便将分区重新分配给另一个成员。默认300000    '''     consumer_1 = KafkaConsumer(        bootstrap_servers=broker_list,        group_id="test000001",        client_id=thread_name,        enable_auto_commit=False,        fetch_min_bytes=1024 * 1024,  # 1M        # fetch_max_bytes=1024 * 1024 * 1024 * 10,        fetch_max_wait_ms=60000,  # 30s        request_timeout_ms=305000,        # consumer_timeout_ms=1,        # max_poll_records=5000,    )    # 设置topic partition    tp = TopicPartition(topic, partition)    # 分配该消费者的TopicPartition，也就是topic和partition，    # 根据参数，每个线程消费者消费一个分区    consumer_1.assign([tp])    # 获取上次消费的最大偏移量    offset = consumer_1.end_offsets([tp])[tp]    print(thread_name, tp, offset)     # 设置消费的偏移量    consumer_1.seek(tp, offset)     print(u"程序首次运行/t线程:", thread_name, u"分区:", partition, u"偏移量:", offset, u"/t开始消费...")     num = 0  # 记录该消费者消费次数    while True:        msg = consumer_1.poll(timeout_ms=60000)        end_offset = consumer_1.end_offsets([tp])[tp]        '''可以自己记录控制消费'''        print(u'已保存的偏移量', consumer_1.committed(tp), u'最新偏移量，', end_offset)        if len(msg) > 0:            print(u"线程:", thread_name, u"分区:", partition, u"最大偏移量:", end_offset, u"有无数据,", len(msg))             lines = 0            for data in msg.values():                for line in data:                    print(line)                    lines += 1                '''                do something                '''            # 线程此批次消息条数             print(thread_name, "lines", lines)            if True:                # 可以自己保存在各topic, partition的偏移量                # 手动提交偏移量 offsets格式：{TopicPartition:OffsetAndMetadata(offset_num,None)}                consumer_1.commit(offsets={tp: (OffsetAndMetadata(end_offset, None))})                if not 0:                    # 系统退出？这个还没试                    os.exit()                    '''                    sys.exit()  只能退出该线程，也就是说其它两个线程正常运行，主程序不退出                    '''            else:                os.exit()        else:            print(thread_name, '没有数据')        num += 1        print(thread_name, "第", num, "次") if __name__ == '__main__':    try:        t1 = MyThread("Thread-0", "test", 0)        threads.append(t1)        t2 = MyThread("Thread-1", "test", 1)        threads.append(t2)        t3 = MyThread("Thread-2", "test", 2)        threads.append(t3)         for t in threads:            t.start()         for t in threads:            t.join()         print("exit program with 0")    except:        print("Error: failed to run consumer program")

# 立刻发送所有数据并等待发送完毕producer.flush() # 读取下一条消息next(consumer) # 手动提交所有已消费的消息consumer.commit() # 手动提交指定的消息consumer.commit([TopicPartition(my_topic, msg.offset)])

import jsonimport tracebackfrom kafka import KafkaProducer, KafkaConsumerfrom kafka.errors import kafka_errors def producer_demo():    # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json    producer = KafkaProducer(        bootstrap_servers=['localhost:9092'],        key_serializer=lambda k: json.dumps(k).encode(),        value_serializer=lambda v: json.dumps(v).encode())    # 发送三条消息    for i in range(0, 3):        future = producer.send(            'kafka_demo',            key='count_num',  # 同一个key值，会被送至同一个分区            value=str(i),            partition=1  # 向分区1发送消息        )        print("send {}".format(str(i)))        try:            future.get(timeout=10)  # 监控是否发送成功                   except kafka_errors:  # 发送失败抛出kafka_errors            traceback.format_exc() def consumer_demo():    consumer = KafkaConsumer(        'kafka_demo',        bootstrap_servers=':9092',        group_id='test'    )    for message in consumer:        print(            f"receive, key: {json.loads(message.key.decode())}, "            f"value: {json.loads(message.value.decode())}"        )

def consumer_demo():    consumer = KafkaConsumer(        'kafka_demo',         bootstrap_servers=':9092',        group_id='test',        enable_auto_commit=False    )    for message in consumer:        print(            f"receive, key: {json.loads(message.key.decode())}, "            f"value: {json.loads(message.value.decode())}"        )        consumer.commit()

consumer = KafkaConsumer(topic, **kwargs)partitions = [TopicPartition(topic, p) for p in consumer.partitions_for_topic(topic)] print("start to cal offset:") # totaltoff = consumer.end_offsets(partitions)toff = [(key.partition, toff[key]) for key in toff.keys()]toff.sort()print("total offset: {}".format(str(toff))) # currentcoff = [(x.partition, consumer.committed(x)) for x in partitions]coff.sort()print("current offset: {}".format(str(coff))) # cal sum and lefttoff_sum = sum([x[1] for x in toff])cur_sum = sum([x[1] for x in coff if x[1] is not None])left_sum = toff_sum - cur_sumprint("kafka left: {}".format(left_sum))