from kafka import KafkaConsumer,TopicPartition
import json
scrapy_kafka_hosts = ["ip:端口", "ip:端口"]
topic = 'wangliang_test'
consumer = KafkaConsumer(bootstrap_servers=scrapy_kafka_hosts,
group_id='', # 消费组
value_deserializer=lambda m: json.loads(m.decode('ascii')), # 消费json 格式的消息
auto_offset_reset='latest', # latest 最新的偏移量，默认最新的 # earliest 最早的偏远量，在还一个组时候才可以使用最早的
enable_auto_commit=True, # 是否开启自动提交 默认开启
auto_commit_interval_ms = 6000 # 提交偏移量的时间间隔，默认5000ms = 5 秒
)
consumer.subscribe([topic]) # 消息的主题，可以指定多个
print(consumer.partitions_for_topic(topic))
# print(consumer.topics()) #获取主题列表
# print(consumer.subscription()) #获取当前消费者订阅的主题
# print(consumer.assignment()) #获取当前消费者topic、分区信息
# print(consumer.beginning_offsets(consumer.assignment())) #获取当前消费者可消费的偏移量 consumer.assign([
TopicPartition(topic=topic, partition=0),
TopicPartition(topic=topic, partition=1),
TopicPartition(topic=topic, partition=3)
]) # 该命令与 subscribe 操作只能存在一个 表示指定主题和分区 获取 consumer 消息 # print(consumer.assignment()) #获取当前消费者topic、分区信息
consumer.seek(TopicPartition(topic=topic, partition=0), 1) # 指定起始offset为12 1表示offset 表示offset 开始连接
# consumer.seek(TopicPartition(topic=topic, partition=1), 0) # 可以注册多个分区，此分区从第一条消息开始接收
consumer.seek(TopicPartition(topic=topic, partition=3), 0) # 没有注册的分区上的消息不会被消费 for msg in consumer:
print(msg)
recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)
print(recv)
print(msg.value["name"],msg.value["age"])
可以指定分区和消费者消费组和管道时间控制，消费数据。
from kafka import KafkaProducer, KafkaConsumer, TopicPartition
import time class ConsumerForKFK(object):
_MESSAGE_NAME = 'wangliang_test' def __init__(self, kafkahost, client_id):
self.kafkaHost = kafkahost
self.group_id = client_id @property
def consumer_client(self, group_id=None):
return KafkaConsumer(
bootstrap_servers=self.kafkaHost,
# client_id = self.client_id,
group_id=self.group_id,
auto_offset_reset="latest"
# 若不指定 consumer_timeout_ms，默认一直循环等待接收，若指定，则超时返回，不再等待
# consumer_timeout_ms ： 毫秒数
# consumer_timeout_ms=5000
) def consumer_seek(self, partition=None, partition_all=None, offset_time=None):
partition_number = range(partition_all) if partition_all else partition
consumer = self.consumer_client
Partition_list = []
for i in partition_number:
Partition_list.append(TopicPartition(topic=self._MESSAGE_NAME, partition=i))
consumer.assign(Partition_list)
print(consumer.assignment()) # 获取当前消费者topic、分区信息
if offset_time:
for Partition in Partition_list:
consumer.seek(Partition, 0)
# 发送到指定的消息主题（异步，不阻塞）
for msg in consumer: # 迭代器，等待下一条消息 print(int(time.time()) - int(msg.timestamp / 1000))
if int(time.time()) - int(msg.timestamp / 1000) <= offset_time: # 打印消息
print(msg)
else:
for msg in consumer: # 迭代器，等待下一条消息
print(msg) scrapy_kafka_hosts = ["ip:端口", "ip:端口"]
topic = 'wangliang_test'
cl = ConsumerForKFK(scrapy_kafka_hosts, "")
cl.consumer_seek(
partition=[1, 2],
# partition_all=3,
offset_time=3000
)