Kafka查询避免OOM

Kafka查询避免OOM

（1）consumer#ThreadLocal中 在拦截器中最后关闭Consumer close

        if(CurrentConsumerHolder.getCurrentConsumer()!=null){
            KafkaConsumer consumer =             
                    CurrentConsumerHolder.getCurrentConsumer();
            consumer.unsubscribe();
            consumer.close(Duration.ofMillis(0));
        }
        CurrentConsumerHolder.clear();

（2）consumer存放到ThreadLocal查询复用

    public KafkaConsumer getQueryConsumer() {
        if (CurrentConsumerHolder.getCurrentConsumer() == null) {
            CurrentConsumerHolder.setCurrentConsumer(
                    new KafkaConsumer<>(this.getKafkaProperties()));
        }
        return CurrentConsumerHolder.getCurrentConsumer();
    }

consumer#poll对象 存放到Redis中增加缓存

提高堆内存

接口限流