官方文件：https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html

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,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執行緒:","分割槽:",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("執行緒:","最大偏移量:",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)
      if is_succeed:
        #更新自己儲存在資料庫中的各topic,partition的偏移量
        is_succeed1 = update_offset(topic,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",1)
    threads.append(t2)
    t3 = MyThread("Thread-2",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")

以上這篇python kafka 多執行緒消費者&手動提交例項就是小編分享給大家的全部內容了，希望能給大家一個參考，也希望大家多多支援我們。