重置偏移量和寻找最新偏移量的无限循环

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='latest',
    enable_auto_commit=False,
    group_id='my_group'
)

while True:
    messages = consumer.poll(timeout_ms=1000)
    for tp, msgs in messages.items():
        for msg in msgs:
            print(f"Received message: {msg.value}")
        consumer.commit()

import hashlib

processed_messages = set()

def is_processed(msg):
    msg_hash = hashlib.sha256(msg.value).hexdigest()
    if msg_hash in processed_messages:
        return True
    processed_messages.add(msg_hash)
    return False

while True:
    messages = consumer.poll(timeout_ms=1000)
    for tp, msgs in messages.items():
        for msg in msgs:
            if not is_processed(msg):
                print(f"Received message: {msg.value}")
                # 处理消息逻辑
        consumer.commit()