在Go语言中，使用Kafka作为消息队列，并将消息写入PostgreSQL数据库是一种常见的应用场景。然而，在这个过程中处理错误是至关重要的。php小编草莓将为您介绍如何在Go中从Kafka读取消息并写入PostgreSQL时，正确地处理错误。通过合理的错误处理，您可以保证应用程序的可靠性和稳定性，提高用户体验。接下来，我们将一步步为您解析这个过程中的错误处理技巧。
问题内容
我正在构建一个 go 应用程序，它从 kafka 主题读取消息并将其写入 postgresql 数据库。
我设置了一个循环，使用 kafka.reader 从 kafka 读取消息，并使用 sql.db 将它们插入数据库。如果读取消息或将其插入数据库时出错，我会记录该错误并继续处理下一条消息。
但是，我不确定如何处理将数据插入 postgresql 数据库后提交 kafka 消息时发生的错误。具体来说，如果手动提交出现错误，该怎么办？我应该重试提交操作吗？我应该记录错误并继续查看下一条消息吗？处理这些类型的错误的最佳实践是什么？
for {
kafkaMessage, err := kafkaReader.ReadMessage(context.Background())
if err != nil {
fmt.Printf("Failed to read message from Kafka: %s\n", err)
continue
}
_, err = db.Exec("INSERT INTO mytable (payload) VALUES ($1)", kafkaMessage.Value)
if err != nil {
fmt.Printf("Failed to insert payload into database: %s\n", err)
continue
}
// What should I do if the commit operation fails?
err = kafkaReader.CommitMessages(context.Background(), kafkaMessage)
if err != nil {
// What's the best practice for handling this error?
}
}
解决方法
当遇到错误时，只需 continue 到 for 循环的下一次迭代。
如果由于任何原因未能提交kafka消息，kafka将在下一次reader.readmessage(ctx)中再次返回相同的消息。
但是为了确保您的代码不会继续徒劳地多次执行相同的失败工作、耗尽资源、用相同的错误消息淹没日志等，请在之后使用简单的 sleep每个错误，或者如果确实需要，请为您的函数使用断路器逻辑。
if err != nil {
log.Errorf("...", ...)
time.Sleep(5 * time.Second)
continue
}
