Kafka查询避免OOM

83 阅读 0 评论 55 点赞

我是靠谱客的博主刻苦饼干，这篇文章主要介绍Kafka查询避免OOM，现在分享给大家，希望可以做个参考。

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

复制代码

1
2
3
4
5
6
7
        if(CurrentConsumerHolder.getCurrentConsumer()!=null){
            KafkaConsumer<String, String> consumer =             
                    CurrentConsumerHolder.getCurrentConsumer();
            consumer.unsubscribe();
            consumer.close(Duration.ofMillis(0));
        }
        CurrentConsumerHolder.clear();

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

复制代码

1
2
3
4
5
6
7
    public KafkaConsumer<String, String> getQueryConsumer() {
        if (CurrentConsumerHolder.getCurrentConsumer() == null) {
            CurrentConsumerHolder.setCurrentConsumer(
                    new KafkaConsumer<>(this.getKafkaProperties()));
        }
        return CurrentConsumerHolder.getCurrentConsumer();
    }

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

提高堆内存

接口限流