如何消费Kafka中的最后一条消息或基于时间戳消费消息？

现有代码的问题梳理 #

• auto.offset.reset配置错误：该参数需设置为字符串"latest"或"earliest"，而非布尔值
• 超时逻辑无效：time.time() > time.time() + timeout的判断永远不成立，需先记录起始时间再做超时校验
• 变量名不一致：定义了data_consumed列表，却使用未声明的kafka_ms进行追加操作
• 缺乏主动定位偏移的逻辑：仅依赖auto.offset.reset无法精准获取最后一条或指定时间戳的消息

解决方案实现 #

1. 消费最后一条消息 #

要获取主题的最后一条消息，需主动定位到每个分区的最新偏移量（需减1，因为最新偏移指向待写入的下一个位置），具体实现如下：

def consume_last_kafka_message(self, timeout=500):
    group_name = "group_name"
    kafka_topic = "your_topic_name"
    config = {
        "bootstrap.servers": "your_broker_server",
        "schema.registry.url": "your_schema_registry_url",
        "group.id": group_name,
        "enable.auto.commit": False,
        "auto.offset.reset": "latest",
        "sasl.mechanisms": "your_sasl_mechanism",
        "security.protocol": "your_security_protocol",
        "sasl.username": "your_username",
        "sasl.password": "your_password"
    }
    
    consumer = AvroConsumer(config)
    data_consumed = []
    # 订阅主题并获取分区信息
    consumer.subscribe([kafka_topic])
    # 确保消费者完成分区分配
    time.sleep(1)
    partitions = consumer.assignment()
    
    if not partitions:
        consumer.close()
        return data_consumed
    
    # 获取每个分区的最新偏移量
    latest_offsets = consumer.end_offsets(partitions)
    
    for partition in partitions:
        latest_offset = latest_offsets[partition]
        if latest_offset > 0:
            # 定位到最后一条消息的位置（最新偏移量-1）
            consumer.seek(partition, latest_offset - 1)
    
    # 拉取最后一条消息
    start_time = time.time()
    while time.time() < start_time + timeout:
        message = consumer.poll(timeout_ms=100)
        if message:
            for _, records in message.items():
                for record in records:
                    data_consumed.append(record.value())
                    # 仅需一条即可退出
                    consumer.close()
                    return data_consumed
    
    consumer.close()
    return data_consumed

2. 基于时间戳消费消息 #

如果要消费指定时间戳之后的所有消息，可通过offsets_for_times获取对应时间戳的偏移量，再定位消费：

def consume_kafka_by_timestamp(self, target_timestamp, timeout=500):
    group_name = "group_name"
    kafka_topic = "your_topic_name"
    config = {
        "bootstrap.servers": "your_broker_server",
        "schema.registry.url": "your_schema_registry_url",
        "group.id": group_name,
        "enable.auto.commit": False,
        "auto.offset.reset": "latest",
        "sasl.mechanisms": "your_sasl_mechanism",
        "security.protocol": "your_security_protocol",
        "sasl.username": "your_username",
        "sasl.password": "your_password"
    }
    
    consumer = AvroConsumer(config)
    data_consumed = []
    consumer.subscribe([kafka_topic])
    time.sleep(1)
    partitions = consumer.assignment()
    
    if not partitions:
        consumer.close()
        return data_consumed
    
    # 构建分区与目标时间戳的映射（时间戳单位为毫秒）
    timestamp_map = {partition: target_timestamp * 1000 for partition in partitions}
    # 获取对应时间戳的偏移量
    offset_results = consumer.offsets_for_times(timestamp_map)
    
    for partition, offset_and_timestamp in offset_results.items():
        if offset_and_timestamp:
            consumer.seek(partition, offset_and_timestamp.offset)
    
    # 开始消费
    start_time = time.time()
    while time.time() < start_time + timeout:
        message = consumer.poll(timeout_ms=100)
        if message:
            for _, records in message.items():
                for record in records:
                    data_consumed.append(record.value())
                    consumer.commit(asynchronous=False)
    
    consumer.close()
    return data_consumed

针对你遇到的问题的修复说明 #

• 问题1：新groupId+latest无返回：通过主动调用end_offsets获取最新偏移并seek到latest_offset-1的位置，即使没有新消息写入，也能拉取到最后一条已存在的消息
• 问题2：旧groupId+latest在Broker重启后异常：主动定位偏移的逻辑不再依赖auto.offset.reset的自动恢复，避免Broker重启后偏移量同步异常导致的消费问题

内容的提问来源于stack exchange，提问作者Xiaoyue Cheng