今天这篇文章《使用 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 流 api
正确答案

我是 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 数据库并返回状态消息》的详细介绍，大家都掌握了吧！