kafka源码分析消费消息

Posted 2021-08-11 顧棟

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了kafka源码分析消费消息相关的知识，希望对你有一定的参考价值。

文章目录

kafka 消费消息源码分析

kafka 消费消息源码分析

消费消息的实例代码

package com.example.demo.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class ConsumerAnalysis {

    public static final AtomicBoolean IS_RUNNING = new AtomicBoolean(true);

    public static Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CommonHelper.BROKER_LIST);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "ConsumerAnalysisGroup-1");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "demo-consumer-client-1");
        return properties;
    }

    public static void main(String[] args) {
        Properties properties = initConfig();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList(CommonHelper.TOPIC));

        try {
            while (IS_RUNNING.get()) {
                ConsumerRecords<String, String> records = consumer.poll(10000);
                System.out.println("records count is " + records.count());
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("topic=" + record.topic()
                            + ", partition = " + record.partition()
                            + ", offset=" + record.offset());

                    System.out.println("key=" + record.offset()
                            + ", value= " + record.value());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }

    }
}

过程步骤

配置消费者客户端参数以及创建消费者实例
订阅Topic
拉取消息并消费
提交消费offset
关闭消费者实例

参数说明

ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG：对应的参数其实就是bootstrap.servers，用于建立到Kafka集群的初始连接的主机/端口对列表。客户端将使用所有服务器，而不管这里为引导指定了哪些服务器;此列表只影响用于发现完整服务器集的初始主机。这个列表的格式应该是host1:port1,host2:port2，…。由于这些服务器仅用于初始连接，以发现完整的集群成员(可能会动态更改)，因此该列表不需要包含完整的服务器集(但是，在服务器关闭的情况下，您可能需要多个服务器)。
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG：对应的参数是key.deserializer，实现org.apache.kafka.common.serialization.Deserializer 接口的密钥的反序列化器类。
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG：对应的参数是value.deserializer，实现 org.apache.kafka.common.serialization.Deserializer 接口的值的反序列化器类。
ConsumerConfig.GROUP_ID_CONFIG：对应的参数是group.id，标识此消费者所属的消费者组的唯一字符串。如果消费者通过使用 subscribe(topic) 或基于 Kafka 的偏移管理策略使用组管理功能，则需要此属性。
ConsumerConfig.CLIENT_ID_CONFIG：对应的参数是client.id，发出请求时传递给服务器的 id 字符串。这样做的目的是通过允许将逻辑应用程序名称包含在服务器端请求日志记录中，从而能够跟踪请求源，而不仅仅是 ip/port。

Note 在创建生产者的时候使用了ConsumerConfig类，在这个类中，是用了static{}块，来初始化一些默认配置。还有一些其他的关于生产者的配置可以在ConsumerConfig类中观察到。

创建消费者实例主流程

补流程图

private KafkaConsumer(ConsumerConfig config,
                      Deserializer<K> keyDeserializer,
                      Deserializer<V> valueDeserializer) {
    try {
        // 如果没有配置客户端id参数，会默认生成一个
        String clientId = config.getString(ConsumerConfig.CLIENT_ID_CONFIG);
        if (clientId.isEmpty())
            clientId = "consumer-" + CONSUMER_CLIENT_ID_SEQUENCE.getAndIncrement();
        this.clientId = clientId;
        // 获取消费者配置
        String groupId = config.getString(ConsumerConfig.GROUP_ID_CONFIG);

        LogContext logContext = new LogContext("[Consumer clientId=" + clientId + ", groupId=" + groupId + "] ");
        this.log = logContext.logger(getClass());

        log.debug("Initializing the Kafka consumer");
        // 配置请求超时的时间，在时间超时且重试次数之后还是没有响应以失败处理
        this.requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
        // 与消费者组进行心跳的超时时间。超过时长则脱离消费组 触发再分配。
        int sessionTimeOutMs = config.getInt(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG);
        // 在没有达到fetch.min.bytes的时候，服务器在响应 fetch 请求之前将阻塞的最长时间。
        int fetchMaxWaitMs = config.getInt(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG);
        // 请求超时时间必须大于心跳的超时时间和阻塞的最长时间
        if (this.requestTimeoutMs <= sessionTimeOutMs || this.requestTimeoutMs <= fetchMaxWaitMs)
            throw new ConfigException(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG + " should be greater than " + ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG + " and " + ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG);
        this.time = Time.SYSTEM;

        // 采集消费者客户端的指标
        Map<String, String> metricsTags = Collections.singletonMap("client-id", clientId);
        MetricConfig metricConfig = new MetricConfig().samples(config.getInt(ConsumerConfig.METRICS_NUM_SAMPLES_CONFIG))
                .timeWindow(config.getLong(ConsumerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
                .recordLevel(Sensor.RecordingLevel.forName(config.getString(ConsumerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
                .tags(metricsTags);
        List<MetricsReporter> reporters = config.getConfiguredInstances(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG,
                MetricsReporter.class);
        reporters.add(new JmxReporter(JMX_PREFIX));
        this.metrics = new Metrics(metricConfig, reporters, time);
        // 请求失败后，重试之间的等待时间
        this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);

        // load interceptors and make sure they get clientId
        // 加载用户配置的拦截器 并输入clientID
        Map<String, Object> userProvidedConfigs = config.originals();
        userProvidedConfigs.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
        List<ConsumerInterceptor<K, V>> interceptorList = (List) (new ConsumerConfig(userProvidedConfigs, false)).getConfiguredInstances(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
                ConsumerInterceptor.class);
        this.interceptors = new ConsumerInterceptors<>(interceptorList);
        // 读取key和vlaue的反序列化器，若用户没有配置，则会忽略这两个配置
        if (keyDeserializer == null) {
            this.keyDeserializer = config.getConfiguredInstance(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                    Deserializer.class);
            this.keyDeserializer.configure(config.originals(), true);
        } else {
            config.ignore(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
            this.keyDeserializer = keyDeserializer;
        }
        if (valueDeserializer == null) {
            this.valueDeserializer = config.getConfiguredInstance(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                    Deserializer.class);
            this.valueDeserializer.configure(config.originals(), false);
        } else {
            config.ignore(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
            this.valueDeserializer = valueDeserializer;
        }
        // 初始化一个集群监听器实例
        ClusterResourceListeners clusterResourceListeners = configureClusterResourceListeners(keyDeserializer, valueDeserializer, reporters, interceptorList);
        // 配置元数据信息，除了异常触发更新，会定时更新元数据
        this.metadata = new Metadata(retryBackoffMs, config.getLong(ConsumerConfig.METADATA_MAX_AGE_CONFIG),
                true, false, clusterResourceListeners);
        // 开始配置服务器地址
        List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
        this.metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), 0);
        String metricGrpPrefix = "consumer";
        ConsumerMetrics metricsRegistry = new ConsumerMetrics(metricsTags.keySet(), "consumer");
        ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config);

        // 配置事务隔离级别
        IsolationLevel isolationLevel = IsolationLevel.valueOf(
                config.getString(ConsumerConfig.ISOLATION_LEVEL_CONFIG).toUpperCase(Locale.ROOT));
        Sensor throttleTimeSensor = Fetcher.throttleTimeSensor(metrics, metricsRegistry.fetcherMetrics);

        // 与消费组的心跳协调时间，该值必须设置为低于 <code>session.timeout.ms</code>，但通常不应设置为高于该值的 1/3。 它可以调整得更低，以控制正常重新平衡的预期时间。
        int heartbeatIntervalMs = config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG);

        // 初始化一个网络客户端
        NetworkClient netClient = new NetworkClient(
                new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, channelBuilder, logContext),
                this.metadata,
                clientId,
                100, // a fixed large enough value will suffice for max in-flight requests
                config.getLong(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG),
                config.getLong(ConsumerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG),
                config.getInt(ConsumerConfig.SEND_BUFFER_CONFIG),
                config.getInt(ConsumerConfig.RECEIVE_BUFFER_CONFIG),
                config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
                time,
                true,
                new ApiVersions(),
                throttleTimeSensor,
                logContext);
        // 将网络客户端转成消费者客户端
        this.client = new ConsumerNetworkClient(
                logContext,
                netClient,
                metadata,
                time,
                retryBackoffMs,
                config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
                heartbeatIntervalMs); //Will avoid blocking an extended period of time to prevent heartbeat thread starvation
        // 配置offset重置策略
        OffsetResetStrategy offsetResetStrategy = OffsetResetStrategy.valueOf(config.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG).toUpperCase(Locale.ROOT));
        // 主题 分区 offset 情况跟踪
        this.subscriptions = new SubscriptionState(offsetResetStrategy);
        // 分配分配器
        this.assignors = config.getConfiguredInstances(
                ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
                PartitionAssignor.class);
        // 创建消费者组协调器实例--里面拥有元数据监听
        this.coordinator = new ConsumerCoordinator(logContext,
                this.client,
                groupId,
                config.getInt(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG),
                config.getInt(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG),
                heartbeatIntervalMs,
                assignors,
                this.metadata,
                this.subscriptions,
                metrics,
                metricGrpPrefix,
                this.time,
                retryBackoffMs,
                config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG),
                config.getInt(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG),
                this.interceptors,
                config.getBoolean(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG),
                config.getBoolean(ConsumerConfig.LEAVE_GROUP_ON_CLOSE_CONFIG));
        // 抓取器 获取与broker之间的线程
        this.fetcher = new Fetcher<>(
                logContext,
                this.client,
                config.getInt(ConsumerConfig.FETCH_MIN_BYTES_CONFIG),
                config.getInt(ConsumerConfig.FETCH_MAX_BYTES_CONFIG),
                config.getInt(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG),
                config.getInt(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG),
                config.getInt(ConsumerConfig.MAX_POLL_RECORDS_CONFIG),
                config.getBoolean(ConsumerConfig.CHECK_CRCS_CONFIG),
                this.keyDeserializer,
                this.valueDeserializer,
                this.metadata,
                this.subscriptions,
                metrics,
                metricsRegistry.fetcherMetrics,
                this.time,
                this.retryBackoffMs,
                this.requestTimeoutMs,
                isolationLevel);

        config.logUnused();
        AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics);

        log.debug("Kafka consumer initialized");
    } catch (Throwable t) {
        // call close methods if internal objects are already constructed
        // this is to prevent resource leak. see KAFKA-2121
        close(0, true);
        // now propagate the exception
        throw new KafkaException("Failed to construct kafka consumer", t);
    }
}

订阅主题

订阅只保留最后一次的Topic，可以是一个，可以是一组，也可以是正则的Topics(会动态变化)。

@Override
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
    // 获取锁和检查消费者是否关闭
    acquireAndEnsureOpen();
    try {
        if (topics == null) {
            throw new IllegalArgumentException("Topic collection to su以上是关于kafka源码分析 消费消息的主要内容，如果未能解决你的问题，请参考以下文章 
 Kafka源码级解析:kafka对消费者分配分区规则
 Flink消费Kafka如何保证相同标识消息的有序性
 一文读懂kafka消息拉取机制｜线程拉取模型
 Kafka源码分析及图解原理之Producer端
 kafka源码分析 生产消息过程
 分享一些 Kafka 消费数据的小经验

kafka源码分析 消费消息

文章目录

kafka 消费消息源码分析

消费消息的实例代码

过程步骤

参数说明

创建消费者实例主流程

订阅主题

kafka源码分析消费消息