在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: %sn", err)
            continue
        }

        _, err = db.Exec("INSERT INTO mytable (payload) VALUES ($1)", kafkaMessage.Value)
        if err != nil {
            fmt.Printf("Failed to insert payload into database: %sn", 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
}

• 关于断路器的文章
• 索尼断路器go 中的实现
• 云原生 go 应用的最佳实践