kafka指定偏移量

在Kafka中，可以通过指定偏移量来消费特定的消息。具体步骤如下：

1. 获取指定 partition 的最新 offset（即最新消息的偏移量）和最早 offset（即最早消息的偏移量）。

2. 决定要从哪个偏移量开始消费消息，可以选择从最新或最早的偏移量开始消费，也可以从指定的偏移量开始消费。

3. 将指定的偏移量作为参数传递给消费者，即可从指定偏移量开始消费消息。

下面是一个示例代码：

from kafka import KafkaConsumer, TopicPartition

# 指定 Kafka broker 的地址
bootstrap_servers = ['localhost:9092']

# 指定要消费的 topic 和 partition
topic = 'example-topic'
partition = 0

# 创建 KafkaConsumer 对象
consumer = KafkaConsumer(bootstrap_servers=bootstrap_servers)

# 获取指定 partition 的最新和最早 offset
latest_offset = consumer.end_offsets([TopicPartition(topic, partition)])[TopicPartition(topic, partition)]
earliest_offset = consumer.beginning_offsets([TopicPartition(topic, partition)])[TopicPartition(topic, partition)]

# 指定要从哪个偏移量开始消费消息
offset = earliest_offset

# 如果要从指定偏移量开始消费消息，可以将 offset 指定为一个数字
# offset = 100

# 将指定偏移量作为参数传递给消费者
consumer.assign([TopicPartition(topic, partition, offset=offset)])

# 开始消费消息
for message in consumer:
    print(message)

在上面的示例代码中，我们首先创建了一个 KafkaConsumer 对象，并指定了要消费的 topic 和 partition。然后，我们获取了指定 partition 的最新和最早 offset，并决定要从哪个偏移量开始消费消息。最后，我们将指定偏移量作为参数传递给消费者，并开始消费消息。