#coding:utf-8
import threading
import os
import sys
from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata
from 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 = 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.assign([tp])
#获取上次消费的最大偏移量
offset = consumer.end_offsets([tp])[tp]
print(thread_name, tp, offset)
# 设置消费的偏移量
consumer.seek(tp, offset)
print u"程序首次运行\t线程:", thread_name, u"分区:", partition, u"偏移量:", offset, u"\t开始消费..."
num = 0 # 记录该消费者消费次数
while True:
msg = consumer.poll(timeout_ms=60000)
end_offset = consumer.end_offsets([tp])[tp]
'''可以自己记录控制消费'''
print u'已保存的偏移量', consumer.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.commit(offsets={tp: (OffsetAndMetadata(end_offset, None))})
if True == 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")
|
