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

)