Kafka 节点关闭时发出警报

Question

我已经公开了一个 Kafka 节点和一个主题名称。我的网络服务器接收到大量的 http 请求数据，我需要对其进行处理，然后将它们推送到 kafka。有时，如果 kafka 节点宕机了，那么我的服务器仍然会不断地抽取数据，这会导致我的内存炸毁，我的服务器也会宕机。

我想要的是在 Kafka 宕机时停止发布数据。我的Java示例代码如下：

  static Producer producer;

  Produce() {
    Properties properties = new Properties();
    properties.put("request.required.acks","1");
    properties.put("bootstrap.servers","localhost:9092,localhost:9093,localhost:9094");
    properties.put("enabled","true");
    properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    properties.put("kafka-topic","pixel-server");
    properties.put("batch.size","1000");
    properties.put("producer.type","async");
    properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
    producer = new KafkaProducer<String, String>(properties);
  }


  public static void main(String[] args) {
    Produce produce = new Produce();

    produce.send(producer, "pixel-server", "Some time");

  }

  //This method is called lot of times
  public void send(Producer<String, String> producer, String topic, String data) {
    ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, data);
    Future<RecordMetadata> response = producer.send(producerRecord, (metadata, exception) -> {
      if (null != exception) {
        exception.printStackTrace();
      } else {
        System.out.println("Done");
      }
    });

我刚刚抽象出了一些示例代码。 send 方法被多次调用。如果卡夫卡关闭，我只想阻止发送任何消息。解决这种情况的有效方法是什么。

Answer 1

如果我是你，我会尝试实现circuit breaker。当您在发送记录时遇到合理数量的故障时，电路会中断并提供一些回退行为。一旦满足某些条件（例如：经过一段时间），电路就会关闭，您将再次发送记录。还有vertx.io 附带它的own solution。