事件驱动架构在现代软件系统中十分常见，它让各组件能够异步通信。传统实现通常借助 Kafka、Google Pub/Sub 或 Amazon SQS 等消息中间件，但在某些场景下，我们可能想用更轻量又足够可靠的方案进行学习或满足定制需求。
本文演示如何基于 Golang + Redis Streams 搭建一个高可靠性的事件驱动系统。
为什么事件驱动系统需要“可靠性”
在很多业务里，丢失事件是不可接受的。以告警系统为例，若漏掉一次关键告警，可能导致宕机、数据泄露或交易失败。因而系统必须满足：
- 持久化（Durability）：事件在被处理前必须保存下来；
- 确认与重试（Ack & Retry）：消费失败不能导致事件丢失；
- 可扩展（Scalability）：支持多生产者、多消费者并发处理。
为什么选 Redis Streams 而非 Pub/Sub？
Redis 原生 Pub/Sub 只做即时推送，订阅者离线时消息直接丢弃；而 Redis Streams 提供：
- 消息持久化；
- Consumer Group，便于水平扩展；
- 消息确认及重放；
- 高效处理大规模事件。
系统架构
- 事件生产者：产生事件并写入 Redis Stream；
- Redis Streams：中央事件存储与消息分发；
- 事件消费者：读取、处理并确认事件。
Golang 实现
1. 启动 Redis
redis-server
2. 安装 Go 客户端
go get github.com/redis/go-redis/v9
3. 事件生产者
package mainimport ("context""fmt""log""github.com/redis/go-redis/v9"
)var ctx = context.Background()func main() {client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})event := map[string]interface{}{"message": "Critical alert! Server down."}_, err := client.XAdd(ctx, &redis.XAddArgs{Stream: "alerts",Values: event,}).Result()if err != nil {log.Fatalf("发布事件失败: %v", err)}fmt.Println("事件发布成功")
}
4. 事件消费者
package mainimport ("context""fmt""log""github.com/redis/go-redis/v9"
)var ctx = context.Background()func main() {client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})for {res, err := client.XRead(ctx, &redis.XReadArgs{Streams: []string{"alerts", "$"}, // "$" 表示从最新位置开始Count: 1,Block: 0, // 阻塞等待}).Result()if err != nil {log.Fatalf("读取事件失败: %v", err)}for _, stream := range res {for _, msg := range stream.Messages {fmt.Printf("处理事件: %v\n", msg.Values)}}}
}
走向生产的强化点
尽管这只是一个简单的演示，真正用于生产的版本还应包含以下功能：
- 错误处理与重试：在失败时实现指数退避重试机制；
- 消费者组：将负载分配给多个消费者以实现并行处理；
- 监控与日志：持续追踪事件处理的各项指标；
- 持久化与备份：启用磁盘持久化，防止数据丢失并支持备份。
结语
借助 Redis Streams + Golang，我们可以构建一个具备持久化、确认机制和水平扩展能力的轻量事件驱动系统，非常适合学习及小型高可用场景。
- 知识星球：云原生AI实战营。10+ 高质量体系课（ Go、云原生、AI Infra）、15+ 实战项目，P8 技术专家助你提高技术天花板，冲击百万年薪！
- 公众号：令飞编程，分享 Go、云原生、AI Infra 相关技术。回复「资料」免费下载 Go、云原生、AI 等学习资料；
- 哔哩哔哩：令飞编程 ，分享技术、职场、面经等，并有免费直播课「云原生AI高新就业课」，大厂级项目实战到大厂面试通关；
