当Kafka Producer不发送消息时如何保存在本地存储消息中？

Question

我有120个数据生产者和Kakfa Producer，他们每秒发送消息，并且其中一些具有敏感数据（我是说，我们必须发送是或是），而且这些生产者有时可能会关闭或失去互联网连接，所以我需要故障生产者的宽容。我知道有多种方式可以同步发送和异步发送，并且它们在不发送数据时会继续尝试发送并将其保存在内存中，但是如果我想将其保存在磁盘中而不是缓冲存储器中，我们该怎么做。如果未发送消息，我们如何做才能将消息保留在磁盘上？我们如何控制卡夫卡生产者何时抛出异常？

Answer 1

您可以在本地事务中将消息插入数据库表。之后，您可以使用线程发送消息，而在从Kafka收到确认后，可以将messages表中的send列更新为true。因此，您的线程应不断从message表中的send字段为false的消息中读取消息，并将消息发送给Kafka。 （如果您有多个实例，领导者实例应负责将消息发送到Kafka以避免重复）

注意：您还可以使用单独的服务向Kafka发送消息，如下所示。

有关更多信息，您可以检查this。

Answer 2

我建议改为立即写入文件。

使用在每台计算机上运行的Filebeat，Fluentbit，Flume等工具将数据集中到Kafka中>]