介绍
在构建分布式系统时，像 amazon sqs 这样的消息队列在处理异步工作负载方面发挥着至关重要的作用。在这篇文章中，我将分享我在 go 中实现强大的 sqs 消费者的经验，该消费者可以处理 keycloak 的用户注册事件。该解决方案使用扇出/扇入并发模式来高效处理消息，而不会占用系统资源。
挑战
我遇到了一个有趣的问题：每天处理大约 50,000 个 sqs 事件以在 keycloak 中注册用户。一种幼稚的方法可能会为每条消息生成一个新的 goroutine，但这可能很快导致资源耗尽。我们需要一种更受控制的并发方法。
为什么要扇出/扇入？
扇出/扇入模式非常适合此用例，因为它：
- 维护固定的工作协程池
- 在工人之间均匀分配工作
- 防止资源耗尽
- 提供对并发操作的更好控制
实施深入探讨
1. 消费者结构
首先我们看一下我们的基本消费结构：
type consumer struct {
client *sqs.client
queuename string
}
2. 消息处理管道
该实现由三个主要组件组成：
- 消息接收者：不断轮询sqs以获取新消息
- 工作池：处理消息的 goroutine 数量固定
- 消息通道：将接收者连接到工作人员
以下是我们启动消费者的方式：
func startpool[requestbody any](
servicefunc func(c context.context, dto *requestbody) error,
consumer *consumer) {
ctx := context.background()
params := &sqs.receivemessageinput{
maxnumberofmessages: 10,
queueurl: aws.string(consumer.queuename),
waittimeseconds: 20,
visibilitytimeout: 30,
messageattributenames: []string{
string(types.queueattributenameall),
},
}
msgch := make(chan types.message)
var wg sync.waitgroup
// start worker pool first
startpool(ctx, msgch, &wg, consumer, servicefunc)
// then start receiving messages
// ... rest of the implementation
}
3. 关键配置参数
让我们检查一下关键的 sqs 配置参数：
全诚外卖通是全诚团队继“全诚商城”之后以叫餐（预订）为核心的又一力作，或者称之为“特色店铺系统”，系统是基于.net2.0 + SQL构架、B/s框架的多用户店铺管理系统；外卖通的开发旨在以商家和消费者为服务对象，借以二者相互依存的关系，以互动的形式成就全诚外卖通一个完善的WEB系统平台，在这个平台里，商家可以扩大销售
- maxnumberofmessages (10)：每次轮询的批量大小
- waittimeseconds (20)：长轮询持续时间
- visibilitytimeout (30)：消息处理的宽限期
4. 工作池实施
工作池是扇出模式发挥作用的地方：
func startpool[requestbody any](
ctx context.context,
msgch chan types.message,
wg *sync.waitgroup,
consumer *consumer,
servicefunc func(c context.context, dto *requestbody) error) {
processingmessages := &sync.map{}
// start 10 workers
for i := 0; i < 10; i++ {
go worker(ctx, msgch, wg, consumer, processingmessages, servicefunc)
}
}
5. 重复消息处理
我们使用sync.map来防止处理重复消息：
if _, loaded := processingMessages.LoadOrStore(*msg.MessageId, true); loaded {
wg.Done()
continue
}
最佳实践和学习
- 错误处理：始终优雅地处理错误并适当记录它们
- 消息清理：仅在成功处理后删除消息
- 优雅关闭：使用上下文实现正确的关闭机制
- 监控：在关键点添加日志记录以提高可观察性
性能考虑因素
- 工作人员数量：根据您的工作负载和可用资源进行选择
- 批量大小：吞吐量和处理时间之间的平衡
- 可见性超时：根据您的平均处理时间设置
未来的改进
- 动态工作人员扩展：根据队列深度调整工作人员数量
- 断路器：为下游服务添加断路器
- metrics collection：添加 prometheus 指标进行监控
- 死信队列：对失败消息实施dlq处理
- 重试：为瞬时失败添加指数退避
结论
扇出/扇入模式为在 go 中处理大量 sqs 消息提供了一个优雅的解决方案。通过维护固定的工作池，我们可以避免无限制的 goroutine 创建的陷阱，同时确保高效的消息处理。
请记住在实现此类模式时始终考虑您的特定用例。此处显示的配置值（工作线程数、超时值等）应根据您的要求和资源限制进行调整。
源代码：[链接到您的存储库（如果有）
标签：#golang #aws #sqs #concurrency #distributed-systems
