我已经公开了一个Kafka节点和一个主题名称。我的网络服务器收到很多http请求数据,我需要先处理它们,然后将它们推送到kafka。有时,如果kafka节点出现故障,那么我的服务器仍会继续抽取数据,这会导致内存消耗and尽,服务器也将崩溃。
我想在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方法被多次调用。我只是想防止在kafka掉线时发送任何消息。解决这种情况的有效方法是什么?
最佳答案
如果您是我,我将尝试实现circuit breaker。当您在发送记录时遇到相当数量的故障时,电路会中断并提供一些后备行为。满足某些条件(例如:经过一段时间)后,电路将关闭,您将再次发送记录。另外,vertx.io
带有own solution。