Golang小白一枚，正在不断学习积累知识，现将学习到的知识记录一下，也是将我的所得分享给大家！而今天这篇文章《使用 Kafka 使用队列处理 csv 文件以插入到 postgres 数据库并返回状态消息》带大家来了解一下##content_title##，希望对大家的知识积累有所帮助，从而弥补自己的不足，助力实战开发！
我正在寻找一些最佳实践/建议来处理 csv 文件，以便使用队列机制（kafka）插入数据库
所以我要做什么：
创建一个新的sql表service request来存储用户请求的信息，例如：
RequestID, Status, Payload, Response如您所见，我有字段 status 来指示请求是成功或失败
这是用户上传 csv 文件时的流程：
- 用户提交 csv 文件
- 验证 csv 文件以确保其使用正确的模板
- 将 csv 文件上传到 google cloud storage，然后使用 requestid 在
service request表上创建一条新记录，负载为 csv 文件的网址 - 读取 csv 文件中的所有记录并将队列发送到 kafka 主题（带有 json 负载）
在消费者方面：
- 监听该主题的所有传入队列（消费队列）
- 处理所有队列
- 如果出现错误，请创建一个 csv 文件来存储此队列失败的原因
- 如果 requestid xxx 的所有队列均已完成，则更新
status并使用 csv 文件错误列表设置响应
所以问题是： 如何知道所有queue的requestid xxx 已全部消耗完，我可以更新状态吗？
我正在使用： go + confluence-kafka-go 库
更新
经过一些研究，我发现它应该通过实现 groupbykey 来使用 kafka stream，在 go 中可以做到这一点吗？我无法从 confluence-kafka-go
正确答案
我是 Kafka 新手，所以我可能不是提供建议的最佳人选，但我最初的反应是强制消息处理“按顺序”发生。在生产者方面，您将指示最后一条消息。在消费者方面，您将读取指示器，一旦到达最后一条消息，您将更新 Status 字段。请记住，强制消息顺序可能会对系统吞吐量产生影响。
有用的阅读可拨打 https://medium.com/latentview-data-services/how-to-use-apache-kafka-to-guarantee-message-ordering-ac2d00da6c22
另一种方法是使用 Postgres 作为分布式锁并跟踪进度。例如，假设您有一个包含以下列的跟踪表：RequestId、RecordsProcessed、RecordsGenerate。每次使用消息时，您都会锁定行或表并递增 RecordsProcessed 列。处理完所有记录后，您将相应地更新 Status。
理论要掌握，实操不能落！以上关于《使用 Kafka 使用队列处理 csv 文件以插入到 postgres 数据库并返回状态消息》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注编程网公众号吧！
