相关疑难解决方法(0)

使用kafka-python-1.0.2。

如果我有一个包含10个分区的主题，那么如何遍历各个分区和消息，同时提交一个特定的分区。我似乎无法在文档中或其他任何地方找到此示例

从文档中，我想使用：

consumer.commit(offset=offsets)

具体来说，如何创建偏移量所需的分区和OffsetAndMetadata字典（字典，可选）– {TopicPartition：OffsetAndMetadata}。

我希望函数调用就像这样：

consumer.commit(partition, offset)

但这似乎并非如此。

提前致谢。

我正在尝试使用kafka-python构建一个应用程序,其中消费者从一系列主题中读取数据.非常重要的是,消费者永远不会两次读取相同的消息,但也永远不会错过消息.

一切似乎工作正常,除非我关闭消费者(例如失败)并尝试从偏移开始阅读.我只能读取主题中的所有消息(创建双读)或仅侦听新消息(并且错过在故障期间发出的消息).暂停消费者时我没有遇到这个问题.

我创建了一个孤立的模拟,以试图解决问题.

这里是通用生产者:

from time import sleep
from json import dumps
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

x=0 # set manually to avoid duplicates 

for e in range(1000):
    if e <= x:
        pass
    else:
        data = dumps(
            {
            'number' : e
        }
        ).encode('utf-8')

        producer.send('numtest', value=data)
        print(e, ' send.')

        sleep(5)

和消费者.如果auto_offset_reset设置为'earliest',则将再次读取所有消息.如果auto_offset_reset设置为'latest',则不会读取停机期间的消息.

from kafka import KafkaConsumer
from pymongo import MongoClient
from json import loads

## Retrieve …