如何以稳健的方式处理 kafka 发布失败

Question

我正在使用 Kafka，我们有一个用例来构建一个容错系统，在这个系统中甚至不会遗漏任何一条消息。所以问题来了： 如果由于任何原因（ZooKeeper 宕机、Kafka 代理宕机等）导致发布到 Kafka 失败，我们如何能够稳健地处理这些消息并在事情再次备份时重播它们。正如我所说，我们甚至无法承受单个消息失败。 另一个用例是，我们还需要在任何给定时间点知道有多少消息由于任何原因未能发布到 Kafka，例如计数器功能，现在这些消息需要重新发布。

其中一个解决方案是将这些消息推送到某个数据库（例如 Cassandra，其中写入速度非常快，但我们还需要计数器功能，我猜 Cassandra 计数器功能不是那么好，我们不想使用它。）它可以处理这种负载，还为我们提供了非常准确的计数器设施。

这个问题更多是从架构的角度来看，然后是使用哪种技术来实现。

PS：我们处理一些像 3000TPS 的地方。因此，当系统开始失败时，这些失败的消息会在很短的时间内快速增长。我们正在使用基于 java 的框架。

感谢您的帮助！

Answer 1

我迟到了。但是我在上面的答案中看到了一些缺失:)

选择像 Cassandra 这样的分布式系统的策略是一个不错的主意。一旦 Kafka 启动并正常，您可以重试写入其中的所有消息。

我想回答“知道在给定时间有多少消息未能发布”

从标签中，我看到您正在使用apache-kafka 和kafka-consumer-api。您可以为您的生产者编写自定义回调，此回调可以告诉您消息是失败还是成功发布。失败时，记录消息的元数据。

现在，您可以使用日志分析工具来分析您的故障。 Splunk 就是这样一个不错的工具。

下面是一个小代码 sn-p，它可以更好地解释我所说的回调：

public class ProduceToKafka {

  private ProducerRecord<String, String> message = null;

 // TracerBulletProducer class has producer properties
  private KafkaProducer<String, String> myProducer = TracerBulletProducer
      .createProducer();

  public void publishMessage(String string) {

    ProducerRecord<String, String> message = new ProducerRecord<>(
        "topicName", string);

    myProducer.send(message, new MyCallback(message.key(), message.value()));
  }

  class MyCallback implements Callback {

    private final String key;
    private final String value;

    public MyCallback(String key, String value) {
      this.key = key;
      this.value = value;
    }


    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
      if (exception == null) {
        log.info("--------> All good !!");
      } else {
        log.info("--------> not so good  !!");
        log.info(metadata.toString());
        log.info("" + metadata.serializedValueSize());
        log.info(exception.getMessage());

      }
    }
  }

}

如果您分析每个时间单位的"--------> not so good !!" 日志数量，您可以获得所需的洞察力。

神速！

Answer 2

Kafka 以分布式、容错方式构建的原因是为了处理与您的问题完全一样的问题，核心组件的多次故障应该避免服务中断。为避免 Zookeeper 宕机，请部署至少 3 个 Zookeeper 实例（如果在 AWS 中，请跨可用区部署它们）。为避免代理失败，请部署多个代理，并确保您在生产者bootstrap.servers 属性中指定了多个代理。要确保 Kafka 集群已将您的消息写入持久庄园，请确保在生产者中设置了 acks=all 属性。当所有同步副本确认收到消息时，这将确认客户端写入（以吞吐量为代价）。您还可以设置排队限制，以确保如果对代理的写入开始备份，您可以捕获异常并处理它并可能重试。

使用 Cassandra（另一个经过深思熟虑的分布式容错系统）来“暂存”您的写入似乎不会为您的架构增加任何可靠性，但确实会增加复杂性，而且 Cassandra 并不是为了消息队列的消息队列，我会避免这种情况。

如果配置正确，Kafka 应该可以处理您的所有消息写入并提供适当的保证。

Answer 3

Chris 已经讲过如何保持系统容错。

Kafka 默认支持at-least once 消息传递语义，这意味着当它尝试发送消息时，会尝试重新发送。

当您创建Kafka Producer 属性时，您可以通过将retries 选项设置为大于0 来进行配置。

 Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:4242");
 props.put("acks", "all");
 props.put("retries", 0);
 props.put("batch.size", 16384);
 props.put("linger.ms", 1);
 props.put("buffer.memory", 33554432);
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

 Producer<String, String> producer = new KafkaProducer<>(props);

更多信息请查看this。