如何在春季云流kafka中存储发送到主题的失败数据

Question

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

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

我正在创建一个获取连续数据流的数据管道。我在运行 3 个代理的 kafka 主题中存储数据流。到目前为止没有问题。我担心假设 3 个代理停止了 5 分钟那时我无法获取有关 kafka 主题的数据。将丢失 5 分钟的数据。从 spring boot 我会收到警告

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，描述为

生产者可用于缓冲等待发送到服务器的记录的总内存字节数。如果记录的发送速度快于它们可以传递到服务器的速度，则生产者将最多阻塞。 block.ms 之后它将引发异常。

此设置应大致对应于生产者将使用的总内存，但不是硬性限制，因为并非生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启用了压缩）以及维护正在进行的请求。

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

如果在一段时间内只有一个代理无法访问，则无需执行任何操作，因为 Kafka 在内部会将主题的分区领导者切换到另一个代理（当然，如果分区被复制了）。