Python-Kafka定时消费：按时间区间消费消息的方案咨询

方案对比：存储时间戳 vs 跟踪Offset #

• 存储结束时间到文件：逻辑简单，但时间戳和Kafka消息的offset并非严格绑定，若生产者时间戳配置错误，会导致消费范围不准确，容易漏/重复消费。
• 跟踪并存储消费Offset：依赖Kafka的offset机制，能精确控制消费区间，避免漏/重复，是更可靠的方案，推荐使用。

核心实现思路 #

每小时执行的脚本需要完成以下步骤：

1. 确定本次要消费的时间窗口（比如脚本在10:00执行，就处理09:00-10:00的消息）
2. 针对Topic的每个分区，通过时间戳找到对应的起始Offset（09:00对应的Offset）和结束Offset（10:00对应的Offset）
3. 从起始Offset开始消费，直到达到结束Offset后停止
4. 将本次的结束Offset存储到本地文件（比如JSON格式），供下次脚本作为起始Offset使用（也可以直接用时间戳找下次的起始点，但存储Offset更精确）

修改后的代码实现 #

import json
import sys
from datetime import datetime, timedelta
from confluent_kafka import Consumer, KafkaError, KafkaException, TopicPartition

def msg_process(msg):
    # 替换为你的消息处理逻辑
    print(f"Received message: {msg.value().decode('utf-8')} at offset {msg.offset()}")

def load_last_offset(file_path):
    # 加载上次存储的结束Offset，无文件则返回空字典
    try:
        with open(file_path, 'r') as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

def save_last_offset(file_path, offsets):
    # 存储本次消费的结束Offset
    with open(file_path, 'w') as f:
        json.dump(offsets, f)

def consume_time_window(consumer, topic, start_time, end_time, offset_file):
    # 获取当前Topic的所有分区
    partitions = consumer.list_topics(topic).topics[topic].partitions.keys()
    topic_partitions = [TopicPartition(topic, p) for p in partitions]
    
    # 转换时间为Kafka需要的毫秒级时间戳
    start_ts = int(start_time.timestamp() * 1000)
    end_ts = int(end_time.timestamp() * 1000)
    
    # 获取每个分区在起始时间点对应的Offset
    for tp in topic_partitions:
        tp.timestamp = (KafkaError.TIMESTAMP_CREATE_TIME, start_ts)
    start_offsets = consumer.offsets_for_times(topic_partitions)
    
    # 获取每个分区在结束时间点对应的Offset
    for tp in topic_partitions:
        tp.timestamp = (KafkaError.TIMESTAMP_CREATE_TIME, end_ts)
    end_offsets = consumer.offsets_for_times(topic_partitions)
    
    # 整理需要消费的分区及对应的起止Offset
    assigned_tps = []
    for tp, start_offset, end_offset in zip(topic_partitions, start_offsets, end_offsets):
        if start_offset is None:
            # 起始时间前无消息，跳过该分区
            continue
        if end_offset is None:
            # 结束时间后无消息，用分区最新Offset作为终点
            consumer.assign([tp])
            consumer.seek_to_end(tp)
            end_offset = tp.offset
        tp.offset = start_offset.offset
        assigned_tps.append((tp, end_offset.offset))
    
    consumer.assign([tp for tp, _ in assigned_tps])
    
    # 开始消费直到所有分区达到结束Offset
    last_offsets = {}
    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                # 检查所有分区是否已消费完毕
                all_done = True
                for tp, end_offset in assigned_tps:
                    current_offset = consumer.position(tp)
                    if current_offset < end_offset:
                        all_done = False
                        break
                if all_done:
                    break
                continue
            
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    raise KafkaException(msg.error())
            else:
                msg_process(msg)
                # 记录当前分区的下一个起始Offset
                last_offsets[str(msg.partition())] = msg.offset() + 1
        
    finally:
        consumer.close()
        # 保存本次消费的终点Offset
        save_last_offset(offset_file, last_offsets)

if __name__ == "__main__":
    # 消费者配置
    conf = {
        'bootstrap.servers': 'your-kafka-broker:9092',
        'group.id': 'hourly-cron-consumer',
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False  # 禁用自动提交，手动控制Offset
    }
    
    consumer = Consumer(conf)
    target_topic = "your-target-topic"
    offset_storage_file = "/path/to/last_offsets.json"
    
    # 计算本次消费的时间窗口（当前整点往前推1小时）
    now = datetime.now()
    end_time = now.replace(minute=0, second=0, microsecond=0)
    start_time = end_time - timedelta(hours=1)
    
    # 若需要基于上次Offset续接，可取消下方注释并调整逻辑
    # last_offsets = load_last_offset(offset_storage_file)
    
    consume_time_window(consumer, target_topic, start_time, end_time, offset_storage_file)

关键注意事项 #

• Cron执行时机：建议在整点后5-10分钟执行脚本（比如10:05），确保Kafka已接收完该小时内的所有消息。
• 时间戳类型：代码使用TIMESTAMP_CREATE_TIME（生产者创建消息的时间），若需用Broker接收时间，替换为TIMESTAMP_LOG_APPEND_TIME。
• Offset存储：用JSON文件记录每个分区的Offset，确保脚本重启或下次执行时能精确续接。
• 异常处理：可根据业务需求添加更多异常捕获逻辑，避免脚本崩溃导致Offset未保存。

内容的提问来源于stack exchange，提问作者Rafa S