如何在spring云流kafka中存储发送到topic的失败数据

Question

我在Kafka中配置了3个代理，运行在不同的端口上，.I正在使用spring cloud stream kafka

brokers:  localhost:9092,localhost:9093,localhost:9094.

我正在创建一个数据管道，以获得连续的数据流.I正在与运行.Till的3个代理在kafka主题中存储数据流现在没有问题.My问题假设假设3个代理停机了5分钟，那么到那时我无法获取有关kafka的数据主题.There将丢失数据5分钟.From spring启动我将收到警告

2020-10-06 11:44:20.840  WARN 2906 --- [ad | producer-2] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-2] Connection to node 0 (/192.168.1.78:9092) could not be established. Broker may not be available. 

有没有一种方法可以在所有代理都关闭时临时存储数据，并在代理重新启动时重新开始从临时存储中恢复写入主题？

Answer 1

您可以利用生产者正在使用的内部缓冲区将数据发送到集群。KafkaProducer有一个隐藏在幕后的队列和一个专门的I/O线程，它实际将数据发送到集群。

结合生产者配置retries (默认设置为0)，您可能希望增加buffer.memory，如下所示

生产者可以用来缓冲等待发送到服务器的记录的内存总字节数。如果发送记录的速度比发送到服务器的速度快，生产者将阻塞for max.block.ms after，并抛出异常。此设置应大致对应于生产者将使用的总内存，但不是硬限制，因为并非生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩(如果启用了压缩)以及用于维护正在进行的请求。

然而，我不认为让生产者自己处理一个完整的集群故障通常是一个好主意。Kafka本身是为了处理单个经纪人的失败而设计的，但如果你的所有经纪人同时无法控制地崩溃，你可能会遇到比丢失单个生产商的一些数据更大的问题。

如果只有一个代理在一段时间内不可访问，那么就没有什么可做的了，因为Kafka会在内部将主题的分区领导者切换到另一个代理(当然，如果分区被复制的话)。