python - 如何在Kafka中处理一次消息，以便服务重新启动时不会处理所有消息

Question

第一次使用 Kafka，我使用微服务架构学习 Kafka，并且正在寻找下一个问题。

每次我重新启动服务时都会处理主题中的所有消息。有没有办法我只能处理这些消息一次，将它们标记为已读或其他什么？

这是我在 Pytho 3 中的代码片段:

class EmailStreamConsumer:
def __init__(self, bootstrap_servers='localhost:9092'):
    self.__bootstrap_servers = bootstrap_servers
    self.__new_emails_consumer = KafkaConsumer('NewEmails', bootstrap_servers=bootstrap_servers,
                                               auto_offset_reset='earliest')
    self.__sent_emails_consumer = KafkaConsumer('SentEmails', bootstrap_servers=bootstrap_servers,
                                                auto_offset_reset='earliest')

def start(self):
    for message in self.__new_emails_consumer:
        value = message.value.decode('utf-8')
        email = json.loads(value)
        self.send_email(email['content'], email['to_email'], email['title'], email['from_email'])
        print("%s:%d:%d: key=%s value=%s" % (
            message.topic, message.partition, message.offset, message.key, message.value))

我希望该服务只发送一次电子邮件。即使服务重新启动也是如此。

Answer 1

我认为您的问题是您的Kafka-Consumer没有GROUP ID

只需添加:

String groupId = "kafka-new-emails";
properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

Your application will start read from the latest email as your consumer group labeled where the last commit you read was. Also, if you have more than one consumer and one of them gets down, consumer group will help you in making a rebalance as to make the consumer that is online to read from the partition that was assigned to the consumer that is down.