将kafka与应用程序分离

我有一个接收大量get请求的应用程序(5分钟内大约250000个请求)。应用程序解析查询参数并发布到kafka。要发布的代码如下:

public class KafkaProcessor {

  private static final String BATCH_SIZE = "batch.size";
  private static final String REQUEST_REQUIRED_ACKS = "request.required.acks";
  private static final String PRODUCER_TYPE = "producer.type";
  private static final String VALUE_SERIALIZER = "value.serializer";
  private static final String KEY_SERIALIZER = "key.serializer";
  private static final String METADATA_BROKER_LIST = "bootstrap.servers";
  private static final String MAX_BLOCK_MS = "max.block.ms";
  private static final String KAFKA_ENABLED = "enabled";

  private static Properties props = new Properties();
  private static KafkaProducer<String, String> producer;
  private static ProducerRecord<String, String> producerRecord;
  private static String topic;


  static {
    boolean isEnabled = Boolean.parseBoolean(ResourceProps.INSTANCE.getKafkaProps(KAFKA_ENABLED));
    if (isEnabled) {
      //Setting up a producer configuration.
      props.put(METADATA_BROKER_LIST, "x.x.x.x:9092,y.y.y.y:9092");
      props.put(KEY_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer");
      props.put(VALUE_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer");
      props.put(PRODUCER_TYPE, "async");
      props.put(REQUEST_REQUIRED_ACKS, "1");
      props.put(BATCH_SIZE, "1000");
      props.put(MAX_BLOCK_MS, "10000");
      producer = new KafkaProducer<>(props);
      topic = "pixel-server";
    }
  }


  private static void publishToKafka(JSONObject data) {
      producerRecord = new ProducerRecord<String, String>(topic, data.toString());
      producer.send(producerRecord, new Callback() {
        @Override public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
          if (exception != null) {
            exception.printStackTrace();
          }
        }
      });
  }
}

我的应用程序托管在aws实例中。kafka服务器也在另一台aws机器中。

但是,如果kafka关闭或kafka由于任何原因响应缓慢,则我的应用程序将冻结,无法进一步处理任何请求。我想知道如何使我的应用程序独立于卡夫卡,也就是说,如果卡夫卡关闭(或响应缓慢),那么它不应该影响我的应用程序。

我尝试了两种方法,比如如果kafka给出了超时,那么计算超时异常的数量并停止发布到kafka,但是由于请求量非常大,所以到了这个时候,我得到了任何超时异常,我的应用程序就会冻结。

任何帮助或指示将不胜感激。

我用的是卡夫卡0.8.2。我的服务器在Vertx。ubuntu中使用的操作系统。ulimit设置为max。