kafka python 多線程，手動提交 指定partion offset

原文：https://blog.csdn.net/xiaofei2017/article/details/80924800

 

#encoding=utf-8
'''
@author: sea
'''
import threading
 
import os
import sys
from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata
 
from consumers.db_util import *
from consumers.json_dispose import *
from collections import OrderedDict
 
 
threads = []
# col_dic, sql_dic = get()
 
 
class MyThread(threading.Thread):
    def __init__(self, thread_name, topic, partition):
        threading.Thread.__init__(self)
        self.thread_name = thread_name
        # self.keyName = keyName
        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 = '172.16.90.63:6667, 172.16.90.58:6667, 172.16.90.59:6667'
    '''
    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 = KafkaConsumer(bootstrap_servers=broker_list,
                             group_id="xiaofesi",
                             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,
                             # max_poll_interval_ms=60000 無該參數
                             )
    #查出數據庫上次保存的offset，此offset已經是上次消費最后一條的offset的offset+1,也就是這次消費的起始位
    dic = get_kafka(topic, partition)
    tp = TopicPartition(topic, partition)
    print(thread_name, tp, dic['offset'])
    #分配該消費者的TopicPartition，也就是topic和partition，根據參數，我是三個消費者，三個線程，每個線程消費者消費一個分區
    consumer.assign([tp])
    #重置此消費者消費的起始位
    consumer.seek(tp, dic['offset'])
    print("程序首次運行\t線程:", thread_name, "分區:", partition, "偏移量:", dic['offset'], "\t開始消費...")
    num=0 #記錄該消費者消費次數
    # end_offset = consumer.end_offsets([tp])[tp]
    # print(end_offset)
    while True:
        args = OrderedDict()
        msg = consumer.poll(timeout_ms=60000)
        end_offset = consumer.end_offsets([tp])[tp]
        print('已保存的偏移量', consumer.committed(tp),'最新偏移量，',end_offset)
        if len(msg) > 0:
            print("線程:", thread_name, "分區:", partition, "最大偏移量:", end_offset, "有無數據,", len(msg))
            lines=0
            for data in msg.values():
                for line in data:
                    lines+=1
                    line = eval(line.value.decode('utf-8'))
                    '''
                    do something
                    '''
            # 線程此批次消息條數
            print(thread_name,"lines",lines)
            #數據保存至數據庫
            
#             is_succeed = save_to_db(args, thread_name)
            is_succeed=True
            if is_succeed:
                #更新自己保存在數據庫中的各topic, partition的偏移量
                is_succeed1 = update_offset(topic, partition, end_offset)
                #手動提交偏移量 offsets格式：{TopicPartition:OffsetAndMetadata(offset_num,None)}
                consumer.commit(offsets={tp:(OffsetAndMetadata(end_offset,None))})
                print(thread_name,"to db suss",num+1)
                if is_succeed1 == 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")