本文介绍一种不依赖外部中间件、仅用 boltdb 与原生 goroutine 即可实现的可靠持久化定时通知调度方案，适用于单实例场景，兼顾简洁性、重启容错与低资源开销。
本文介绍一种不依赖外部中间件、仅用 boltdb 与原生 goroutine 即可实现的可靠持久化定时通知调度方案，适用于单实例场景，兼顾简洁性、重启容错与低资源开销。
在构建轻量级通知服务（如延时消息、预约提醒）时，核心挑战在于：任务必须在指定时间精确触发，且服务重启后仍能恢复未执行任务。由于题目明确限定“仅需单实例”且已选用 BoltDB（嵌入式、ACID 安全的 key/value 存储），我们应避免引入 RabbitMQ、Redis 或复杂调度框架（如 AGScheduler），转而采用类 Unix cron 的精简内存+持久化协同模型——既保证可靠性，又保持代码清晰可控。
核心设计思路：内存队列 + 时间轮驱动 + 持久化同步
不同于为每个任务启动独立 goroutine（易导致 goroutine 泄漏、无法统一管理），推荐采用 “单调度循环 + 有序待办列表” 架构：
- 启动时从 BoltDB 加载所有 isSent = false 的通知，按 delayUntil 升序排序；
- 维护一个内存中的 []Notification 列表（即“待调度队列”），始终保证首项为最早待触发任务；
- 主调度协程休眠至首个任务的 delayUntil 时间点，唤醒后批量处理所有已到期任务（含并发发送、状态更新、DB 清理）；
- 新增通知通过 API 写入 BoltDB 后，同步插入内存队列并重排序，确保调度器及时感知。
该模型规避了 goroutine 泄漏风险，天然支持重启恢复，且无外部依赖，非常适合中小规模业务场景。
Golang Spf13 Viper
下载
Go 配置库，使用 spf13/viper — 分层优先级（flag > env >file > KV > default），提供 BindPFlag/BindPFlags、SetEnvPrefix + SetEnvKeyReplace 等功能。
示例实现（关键逻辑）
type Notification struct {
ID string `json:"id"`
DelayUntil time.Time `json:"delayUntil"`
User string `json:"user"`
Msg string `json:"msg"`
IsSent bool `json:"isSent"`
}
var pending []Notification // 内存中按 DelayUntil 升序排列的待处理列表
var mu sync.RWMutex
// 启动时加载并初始化调度器
func initScheduler(db *bolt.DB) {
db.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("notifications"))
if b == nil {
return nil
}
b.ForEach(func(k, v []byte) error {
var n Notification
if err := json.Unmarshal(v, &n); err != nil {
return nil // 跳过损坏数据
}
if !n.IsSent && !n.DelayUntil.After(time.Now()) {
// 已过期但未发送 → 立即触发（补偿逻辑）
go sendAndMarkSent(db, n)
} else if !n.IsSent {
pending = append(pending, n)
}
return nil
})
return nil
})
// 按 delayUntil 排序
sort.Slice(pending, func(i, j int) bool {
return pending[i].DelayUntil.Before(pending[j].DelayUntil)
})
// 启动主调度循环
go runSchedulerLoop(db)
}
func runSchedulerLoop(db *bolt.DB) {
for {
mu.RLock()
if len(pending) == 0 {
mu.RUnlock()
time.Sleep(30 * time.Second) // 空闲时降频轮询
continue
}
nextTime := pending[0].DelayUntil
mu.RUnlock()
// 休眠至下一个触发点
sleepDur := time.Until(nextTime)
if sleepDur > 0 {
time.Sleep(sleepDur)
}
// 批量处理所有已到期任务
now := time.Now()
var toProcess []Notification
mu.Lock()
for len(pending) > 0 && !pending[0].DelayUntil.After(now) {
toProcess = append(toProcess, pending[0])
pending = pending[1:]
}
mu.Unlock()
for _, n := range toProcess {
go sendAndMarkSent(db, n)
}
}
}
func sendAndMarkSent(db *bolt.DB, n Notification) {
// 1. 发送通知（HTTP/Email/SMS等）
err := sendNotification(n.User, n.Msg)
if err != nil {
slog.Error("failed to send notification", "id", n.ID, "err", err)
return
}
// 2. 原子更新 DB 标记为已发送
err = db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte("notifications"))
data, _ := json.Marshal(Notification{
ID: n.ID,
DelayUntil: n.DelayUntil,
User: n.User,
Msg: n.Msg,
IsSent: true,
})
return b.Put([]byte(n.ID), data)
})
if err != nil {
slog.Error("failed to update notification status", "id", n.ID, "err", err)
}
}注意事项与增强建议
- ✅ 事务安全：BoltDB 的 Update() 是原子写入，确保 send → markSent 不会因崩溃中断；若发送成功但 DB 更新失败，可通过定期扫描 isSent=false AND delayUntil < now 进行补偿。
- ⚠️ 并发插入处理：API 接收新通知时，需加写锁插入内存队列并重排序（或使用 heap.Interface 维护最小堆提升性能）。
- ? 扩展性提示：若未来需多实例，必须引入分布式锁（如 Redis SETNX）或改用支持事务的队列（如 PostgreSQL SELECT ... FOR UPDATE SKIP LOCKED），此时本文方案即为良好演进基础。
- ? 精度权衡：time.Sleep() 在毫秒级精度足够，若需亚秒级强实时，可结合 time.Ticker + 微调逻辑，但对通知类场景通常非必需。
此方案以约 150 行核心代码，实现了生产就绪的持久化定时调度能力——无第三方依赖、重启自愈、逻辑透明、易于监控与调试，是 Go 生态中“恰到好处”的工程实践范例。
立即学习“go语言免费学习笔记（深入）”；
