java - kafka消费者内存消耗高

标签 java apache-kafka spring-kafka

最近我发现kafka消费者需要大量的内存。

对于测试，我刚刚在本地启动了一个监听单个主题的单线程消费者。主题有 4 个分区。 Kafka 只有一个经纪人。

我只从制作人那里发送了 10 条小消息(当时是晚上 11:44:30 左右，请参阅我在链接中附加的图片)。从那时起，没有人再向该主题发送任何消息。 从那时起，我一直在图表上看到消费者轮询工作期间内存消耗不断增长。线路不断增长，直到 GC 未被调用。 消费者只发送轮询请求，不返回任何内容，但需要大量内存。

我认为这是一个问题。 我尝试进行一些调整，即将一​​些参数配置为 FETCH_MAX_BYTES_CONFIG/MAX_PARTITION_FETCH_BYTES_CONFIG/MAX_POLL_RECORDS_CONFIG 但实际上没有任何结果。

SSCCE:

Kafka消费者配置:

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, byte[]> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, byte[]> factory
                = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConcurrency(1);
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        return factory;
    }

    private ConsumerFactory<String, byte[]> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProperties());
    }

    private Map<String, Object> consumerProperties() {
        final Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "local-test-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        //props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1000000); //1mb
        //props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
        //props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 256000); //256kb
        return props;
    }

}

Kafka数据监听器

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class KafkaDataListener {

    @KafkaListener(topics = "local-test-topic", containerFactory = "kafkaListenerContainerFactory")
    public void consumeEvent(byte[] eventData, final Acknowledgment ack) {
         try {
             System.out.println("consumer received message:" + Arrays.toString(eventData));
         }finally {
             ack.acknowledge();
         }
    }
}

主应用程序

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class App {
    public static void main(String[] args) {
        SpringApplication.run(App.class, args);
    }
}

构建.gradle:

group 'org.test.kafka'
version '1.0-SNAPSHOT'

apply plugin: 'java'

sourceCompatibility = 1.8

repositories {
    mavenLocal()
    mavenCentral()
}

dependencies {

    compile group: 'org.springframework.boot', name: 'spring-boot-starter', version: '1.5.4.RELEASE'
    compile group: 'org.springframework.kafka', name: 'spring-kafka', version: '1.2.2.RELEASE'
}

输出:

2018-02-02 23:38:51.506  INFO 14888 --- [           main] org.kafka.tests.App                      : Starting App on so-workstation with PID 14888 (/home/.../projects/custom/kafka-consumer-test/out/production/classes started by ... in /home/.../projects/custom/kafka-consumer-test)
2018-02-02 23:38:51.508  INFO 14888 --- [           main] org.kafka.tests.App                      : No active profile set, falling back to default profiles: default
2018-02-02 23:38:51.591  INFO 14888 --- [           main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@281e3708: startup date [Fri Feb 02 23:38:51 MSK 2018]; root of context hierarchy
2018-02-02 23:38:52.028  INFO 14888 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.kafka.annotation.KafkaBootstrapConfiguration' of type [org.springframework.kafka.annotation.KafkaBootstrapConfiguration$$EnhancerBySpringCGLIB$$2d249aa5] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2018-02-02 23:38:52.205  INFO 14888 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
2018-02-02 23:38:52.207  INFO 14888 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 0
2018-02-02 23:38:52.221  INFO 14888 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
    auto.commit.interval.ms = 5000
    auto.offset.reset = latest
    bootstrap.servers = [localhost:9092]
    check.crcs = true
    client.id = 
    connections.max.idle.ms = 540000
    enable.auto.commit = false
    exclude.internal.topics = true
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    group.id = local-test-consumer-group
    heartbeat.interval.ms = 3000
    interceptor.classes = null
    key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.ms = 50
    request.timeout.ms = 305000
    retry.backoff.ms = 100
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    session.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer

2018-02-02 23:38:52.225  INFO 14888 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
    auto.commit.interval.ms = 5000
    auto.offset.reset = latest
    bootstrap.servers = [localhost:9092]
    check.crcs = true
    client.id = consumer-1
    connections.max.idle.ms = 540000
    enable.auto.commit = false
    exclude.internal.topics = true
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    group.id = local-test-consumer-group
    heartbeat.interval.ms = 3000
    interceptor.classes = null
    key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.ms = 50
    request.timeout.ms = 305000
    retry.backoff.ms = 100
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    session.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer

2018-02-02 23:38:52.258  INFO 14888 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.10.2.0
2018-02-02 23:38:52.258  INFO 14888 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : 576d93a8dc0cf421
2018-02-02 23:38:52.268  INFO 14888 --- [           main] org.kafka.tests.App                      : Started App in 1.056 seconds (JVM running for 1.337)
2018-02-02 23:38:52.308  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : Discovered coordinator kafka:9092 (id: 2147483646 rack: null) for group local-test-consumer-group.
2018-02-02 23:38:52.312  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : Revoking previously assigned partitions [] for group local-test-consumer-group
2018-02-02 23:38:52.313  INFO 14888 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked:[]
2018-02-02 23:38:52.313  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : (Re-)joining group local-test-consumer-group
2018-02-02 23:38:52.319  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : Successfully joined group local-test-consumer-group with generation 1
2018-02-02 23:38:52.320  INFO 14888 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : Setting newly assigned partitions [local-test-topic-3, local-test-topic-2, local-test-topic-1, local-test-topic-0] for group local-test-consumer-group
2018-02-02 23:38:52.328  INFO 14888 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned:[local-test-topic-3, local-test-topic-2, local-test-topic-1, local-test-topic-0]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]
consumer received message:[...]

有人知道如何正确调整它吗？ 或者也许在某些更高版本上优化了内存使用？

卡夫卡服务器:0.10.2.0 卡夫卡客户端:0.10.2.0

查看图片:

kafka 0.10.2 memory consumption diagram kafka 0.10.2 memory allocation per thread diagram

更新:

对于kafka消费者1.0.0(spring kafka 2.1.2)内存使用图看起来更好一些。现在消费线的增长速度没有以前那么快了。 但现在 RMI TCP 连接线程消耗的内存甚至比 kafka 消费者线程还要多。

kafka 1.0.0 memory consumption diagram with default settings kafka 1.0.0 memory allocation per thread diagram with default settings

此外，消费者的参数似乎正在影响内存使用。 使用消费者参数 FETCH_MAX_BYTES_CONFIG = 1mb 和 MAX_PARTITION_FETCH_BYTES_CONFIG = 256kb 消耗会降低。

kafka 1.0.0 memory consumption diagram with custom settings

最佳答案

造成这种“锯齿模式”的原因之一是应用程序 Java VisualVM 本身。它每秒都会向您的 JVM 询问信息。然后，JVM 为此进程创建大量对象，这些对象在发送到 VisualVM 后就会过时，因此很容易被垃圾收集。

尝试降低 VisualVM 设置中的轮询率。它应该最大限度地减少影响。

关于java - kafka消费者内存消耗高，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/48591163/

上一篇：php - 从外部应用程序访问 Symfony session 值

下一篇：Java流: group and sort by a previous mapping result?

相关文章：

java - 使用 vlcj 流式传输视频文件

java - Storm Bolt 无法从 Spout 反序列化对象

java - Spring/Avro - 使用融合模式注册表

java - 使用Spring Boot配置Kafka的问题

spring - Kafka监听器接收List<ConsumerRecord<String, String>>，是否可以消费？

java - 创建 Java TCP 协议(protocol)

java - 使用Spring框架是否可以根据某些业务逻辑连接两个不同的数据库

java - (设计思路)唤醒多进程做工作流的最佳实践

apache-kafka - 卡夫卡流并发？

apache-spark - 当Kafka客户端异步发送消息时，Spark流任务正常关闭

©2024 IT工具网  联系我们