etf1/kafka-message-scheduler

Kafka消息调度器

概述

Kafka消息调度器允许在特定时间将消息发送到目标主题，支持指定payload，适用于基于Kafka消息的定时触发事件处理。通过Kafka主题存储调度任务，结合高可用和故障转移机制，确保定时任务可靠执行。

使用场景

定时内容管理

如TF1网站中，视频在指定日期上线和下线，通过调度器发送消息到特定主题，由消费者执行激活/停用操作。

重试机制

处理外部API调用失败时，可将失败消息重新调度以重试，需基于原始时间戳确保幂等性。

用户生命周期管理

用户登录时调度"1年后删除"事件，若用户长期未登录，触发消息以删除不活跃用户。

工作原理

调度器通过Kafka主题存储调度任务，消息需包含以下特定 headers：

• scheduler-epoch：调度时间戳（自1970年起的秒数）
• scheduler-target-topic：目标主题名称
• scheduler-target-key：发送到目标主题的消息键

示例： 调度消息：

Headers:
    scheduler-epoch: 1893456000  # 目标时间戳
    scheduler-target-topic: online-videos  # 目标主题
    scheduler-target-key: vid1  # 目标消息键
    customer-header: dummy  # 自定义header
Timestamp: 1607918336  # 原始消息时间戳
Key: vid1-online  # 调度消息键
Value: "video 1"  # payload

触发后在online-videos主题生成的消息：

Headers:
    scheduler-timestamp: 1607918336  # 原始消息时间戳
    scheduler-key: vid1-online  # 原始调度消息键
    scheduler-topic: schedules  # 原始调度主题
    customer-header: dummy  # 保留自定义header
Key: vid1  # 目标消息键
Value: "video 1"  # 原始payload

触发后，消息会复制到history主题用于审计，并在调度主题生成墓碑消息（空payload）删除该调度。

对于Go语言，可使用clientlib包装Kafka消息以简化调度任务创建。

核心特性

高可用性

支持多实例部署，调度任务自动负载均衡，降低单实例内存压力。可配置多个输入主题，但所有主题中的调度任务键需唯一。

故障转移

调度器崩溃或长时间下线后，重启时会重新同步分配的调度任务，优先触发错过的任务，再处理实时传入的调度。

可观测性

默认在8001端口的/metrics路径暴露Prometheus指标，指标kafka_scheduler_event_total包含：

• 错过的调度数
• 无效的调度数
• 删除的调度数
• 计划中的调度数
• 触发的调度数

配置参数

通过环境变量配置，参数说明如下：

使用方法

Docker部署

bash
docker run -e BOOTSTRAP_SERVERS="kafka:9092" etf1/kafka-message-scheduler

源码运行（Go）

bash
BOOTSTRAP_SERVERS="kafka:9092" go run ./cmd/kafka

集成到Go程序

go
import (
    runner "github.com/etf1/kafka-message-scheduler/runner/kafka"
)

// ...

kafkaRunner := runner.DefaultRunner()
go func() {
    if err := kafkaRunner.Start(); err != nil {
        log.Printf("调度器启动失败: %v", err)
    }
}()

// ...

kafkaRunner.Close()

调度主题创建

调度主题需配置为紧凑模式且保留期无限，分区数至少与调度器实例数相同：

bash
kafka-topics --bootstrap-server "${BOOTSTRAP_SERVERS}" --create --topic schedules \
             --partitions 3 --config "cleanup.policy=compact" --config "retention.ms=-1"