kafka-python文檔

一、consumer

1. 常用api

#建立連接
consumer = KafkaConsumer(bootstrap_servers=['ip1:port','ip2:port'],
                         api_version=(0,10),group_id='my_group')

# topic所有的partition
consumer.partitions_for_topic(topic)

# 構(gòu)造topicPartition對象
tps = [TopicPartition(topic, p) for p in consumer.partitions_for_topic(topic)]

# 為consumer分配分區(qū)
consumer.assign(tps)

# kafka每個分區(qū)的最新offset
consumer.end_offsets(tps)

# 當前groupid 每個分區(qū)消費到的位置
for i in range(len(tps)):
    consumer.position(tps[i])

# 消費數(shù)據(jù)
for message in consumer:
    partition = message.partition
    offset = message.offset
    value = message.value

# 重置offset
for i in range(len(tps)):
    consumer.seek(tps[i], partition_offset[i])    #partition_offset保存每
# partition_offset保存每個分區(qū)的起始消費位置
# 形如{0:123, 1:345 }，表示0分區(qū)從123開始再次消費

二、producer

三、其他

3.1 json處理

額外的包：  
pip install msgpack
import msgpack
producer:
producer = KafkaProducer(value_serializer=msgpack.dumps)
producer.send('msgpack-topic', {'key': 'value'})
consumer:
KafkaConsumer(value_deserializer=msgpack.unpackb)
此時得到的value是dict類型