books

《Java 消息中间件·Kafka 内核与客户端》8.1 KafkaConsumer 的两套实现

KafkaConsumer 在 4.3.0 里只是一个门面，真正的实现是 ClassicKafkaConsumer 或 AsyncKafkaConsumer，分支点在 ConsumerDelegateCreator 而不是 KafkaConsumer 本身。本节用源码和三次运行把默认值、判定逻辑、以及为什么 getClass() 永远测不出真实实现讲清楚。

Leeting Yan 10 分钟阅读 4874 字

一个看似简单的问题

你写下一行 new KafkaConsumer<>(props)，得到的对象到底是什么类？

直觉会答 KafkaConsumer。但如果你读过 KIP-848 的材料，会记得 4.x 引入了新的消费者实现 AsyncKafkaConsumer，于是又会猜「可能 new 出来的是它」。

两个答案都不完整。真实情况是：KafkaConsumer 是一个门面（facade），它自己不实现任何消费逻辑，只持有一个 delegate 字段，把每次调用转给 ClassicKafkaConsumer 或 AsyncKafkaConsumer。建成哪一套，取决于配置项 group.protocol。

这一节要解决三个问题：

  1. 门面与实现的关系在源码里是怎么写的；
  2. 不配置时默认走哪一套；
  3. 怎么在运行时确认自己拿到的是哪一套。

第三个问题有一个反直觉的答案：用 getClass() 是测不出来的。本节会把实测输出贴出来，说明为什么。

门面与委托：三个字段

先看 KafkaConsumer 的类声明附近。文件
org/apache/kafka/clients/consumer/KafkaConsumer.java，第 539–543 行：

public class KafkaConsumer<K, V> implements Consumer<K, V> {

    private static final ConsumerDelegateCreator CREATOR = new ConsumerDelegateCreator();

    private final ConsumerDelegate<K, V> delegate;

三行代码说明了整个结构：

  • KafkaConsumer 实现的是公开接口 Consumer<K, V>，这是使用者唯一应该依赖的类型；
  • 它内部持有的是 ConsumerDelegate<K, V>，这是一个内部接口，ClassicKafkaConsumer 和 AsyncKafkaConsumer 都实现它；
  • 创建委托实例的动作交给静态字段 CREATOR。

构造函数最后只做一件事。同一个文件里：

KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
    delegate = CREATOR.create(config, keyDeserializer, valueDeserializer);
}

所以 KafkaConsumer 的每一个公开方法，本质上都是 delegate.xxx(...)。门面自己不保存 clientId、不持有网络连接、也没有 poll 的主循环。这些全在委托对象里。

这也解释了一个常见困惑：为什么在 KafkaConsumer 的源码里搜 poll 的业务逻辑搜不到，只搜到一行转发。因为主循环在 ClassicKafkaConsumer.poll(Timer) 和 AsyncKafkaConsumer.poll(Duration) 里，两者写法完全不同。

分支在 ConsumerDelegateCreator 里

真正的判定不在 KafkaConsumer，而在 ConsumerDelegateCreator.create。
org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java，第 57–72 行：

public <K, V> ConsumerDelegate<K, V> create(ConsumerConfig config,
                                            Deserializer<K> keyDeserializer,
                                            Deserializer<V> valueDeserializer) {
    try {
        GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));

        if (groupProtocol == GroupProtocol.CONSUMER)
            return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer, Optional.empty());
        else
            return new ClassicKafkaConsumer<>(config, keyDeserializer, valueDeserializer);
    } catch (KafkaException e) {
        throw e;
    } catch (Throwable t) {
        throw new KafkaException("Failed to construct Kafka consumer", t);
    }
}

逐行读：

  • config.getString(GROUP_PROTOCOL_CONFIG) 读的是 group.protocol；
  • .toUpperCase(Locale.ROOT) 说明大小写不敏感，Consumer、CONSUMER、consumer 等价；
  • GroupProtocol.valueOf(...) 只接受 CLASSIC 和 CONSUMER 两个枚举值，写别的会在这一步抛 IllegalArgumentException，随后被包成 KafkaException；
  • 判定条件只有一个：等于 CONSUMER 就走 AsyncKafkaConsumer，其余全部走 ClassicKafkaConsumer。

注意这个 else 的覆盖范围。它不是「等于 CLASSIC 就走经典实现」，而是「不等于 CONSUMER 就走经典实现」。这两句话在默认值存在时是等价的，但在异常输入下的表现不同：任何非法值都不会悄悄落到经典实现，而是在 valueOf 就失败。

文档与实现的一处措辞差异

ConsumerDelegateCreator 的类注释写的是：

This is based on the presence and value of the group.protocol configuration. If the value is present and equal to "consumer", the AsyncKafkaConsumer will be returned.

「present and value」读起来像是「没配就是另一套」。但实现用的是 config.getString(...)，而 ConsumerConfig 给这个配置定义了默认值。所以「没配」并不特殊，它只是拿到了默认值。

以源码为准：判定与配置是否显式出现无关，只看最终值。

默认走哪套：classic

ConsumerConfig.java 第 114–117 行：

public static final String GROUP_PROTOCOL_CONFIG = "group.protocol";
public static final String DEFAULT_GROUP_PROTOCOL = GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT);
public static final String GROUP_PROTOCOL_DOC = "The group protocol that the consumer uses. The " +
    "supported values are <code>classic</code> or <code>consumer</code>. The default value is <code>classic</code>.";

同一文件第 653–658 行把它注册进 ConfigDef：

.define(GROUP_PROTOCOL_CONFIG,
        Type.STRING,
        DEFAULT_GROUP_PROTOCOL,
        ConfigDef.CaseInsensitiveValidString.in(Utils.enumOptions(GroupProtocol.class)),
        Importance.HIGH,
        GROUP_PROTOCOL_DOC)

默认值是 classic。所以 4.3.0 上什么都不配，拿到的是 ClassicKafkaConsumer。

这一点值得强调，因为网上不少文章在 KIP-848 落地后写成「4.x 默认已经切到新协议」。至少在这个版本不是。AsyncKafkaConsumer 的类注释也只说「is intended to be the default in coming releases」——是将来，不是现在。

实测：getClass() 测不出真实实现

按最自然的写法，先 new 一个 consumer，再打印 getClass().getName()。程序如下（ImplClasses.java）：

import java.util.Properties;
import java.lang.reflect.Field;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class ImplClasses {
    private static KafkaConsumer<String, String> make(String gp) {
        Properties p = new Properties();
        p.put("bootstrap.servers", "127.0.0.1:19213");
        p.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        p.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        if (gp != null) p.put("group.protocol", gp);
        return new KafkaConsumer<>(p);
    }

    private static String delegateOf(Object consumer) throws Exception {
        Class<?> c = consumer.getClass();
        while (c != null) {
            for (Field f : c.getDeclaredFields()) {
                if (f.getName().equals("delegate")) {
                    f.setAccessible(true);
                    Object d = f.get(consumer);
                    return d == null ? "null" : d.getClass().getName();
                }
            }
            c = c.getSuperclass();
        }
        return "<no delegate field found>";
    }

    public static void main(String[] args) throws Exception {
        String[] labels = {"默认（不设 group.protocol）", "group.protocol=classic", "group.protocol=consumer"};
        String[] values = {null, "classic", "consumer"};
        for (int i = 0; i < labels.length; i++) {
            try (KafkaConsumer<String, String> c = make(values[i])) {
                System.out.println(labels[i]
                        + " | getClass()=" + c.getClass().getName()
                        + " | delegate=" + delegateOf(c));
            }
        }
    }
}

编译与运行命令：

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
CP="$HOME/.m2/repository/org/apache/kafka/kafka-clients/4.3.0/kafka-clients-4.3.0.jar"
CP="$CP:$HOME/.m2/repository/org/slf4j/slf4j-api/2.0.20/slf4j-api-2.0.20.jar"
$JAVA_HOME/bin/javac -cp "$CP" -d . ImplClasses.java
$JAVA_HOME/bin/java -cp "$CP:." ImplClasses

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

默认（不设 group.protocol） | getClass()=org.apache.kafka.clients.consumer.KafkaConsumer | delegate=org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer
group.protocol=classic | getClass()=org.apache.kafka.clients.consumer.KafkaConsumer | delegate=org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer
group.protocol=consumer | getClass()=org.apache.kafka.clients.consumer.KafkaConsumer | delegate=org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer

三行输出，两个结论：

  1. getClass() 恒为 org.apache.kafka.clients.consumer.KafkaConsumer，三种配置下完全一样。因为 new 出来的确实就是门面对象，委托对象只是它的私有字段。
  2. 真实实现要靠 delegate 字段才能看到，三次分别是 Classic、Classic、Async，与源码判定一致。

这就是本节开头说的反直觉答案。很多资料写「创建后打印类名即可确认实现」，那在 4.3.0 上不成立。如果只打印 getClass()，你会得出「三套配置都一样」的错误结论。

需要说明：上面用到的反射只是观测手段，不是推荐做法。它读取私有字段，在任何一次内部重构后都可能失效。生产代码不应该这样判断实现。

不反射怎么判断

那在真实项目里，怎么确认自己跑在哪套实现上？有三个可观测的信号。

一、构造期的配置校验

两套实现对配置的接受范围不同。ConsumerConfig 第 398–413 行定义了两张黑名单：

private static final List<String> CLASSIC_PROTOCOL_UNSUPPORTED_CONFIGS = List.of(
        GROUP_REMOTE_ASSIGNOR_CONFIG,
        SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
        SHARE_ACQUIRE_MODE_CONFIG
);

private static final List<String> CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS = List.of(
        PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
        HEARTBEAT_INTERVAL_MS_CONFIG,
        SESSION_TIMEOUT_MS_CONFIG,
        SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
        SHARE_ACQUIRE_MODE_CONFIG
);

也就是说，group.protocol=consumer 时不能再显式设置 partition.assignment.strategy、heartbeat.interval.ms、session.timeout.ms——这些在经典协议里由客户端协商，在新协议里由服务端接管。

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29），同时设置 group.protocol=consumer 和 partition.assignment.strategy：

org.apache.kafka.common.config.ConfigException: partition.assignment.strategy cannot be set when group.protocol=CONSUMER

这条异常本身就是证据：能抛出它，说明走的是新协议分支的校验。

二、日志里的实现类名

两套实现的日志 logger 名不同。开启 debug 后可以看到：

  • 经典实现打印 org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer；
  • 新实现打印 org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer。

这比武断地反射私有字段稳，但要求日志级别足够低，且日志框架配置正确。

三、行为差异

最可靠的是行为。例如新协议下客户端不再自己算分配，ConsumerPartitionAssignor 相关配置被禁用；再比如再平衡回调的触发线程不同。6.3 节已经详细对比过两套协议，这里只强调一点：行为差异比类名更能说明问题，因为类名是内部实现，行为是契约。

两套实现的架构差异

为什么要有两套？因为线程模型根本不同。

ClassicKafkaConsumer.java 第 103–105 行：

/**
 * A client that consumes records from a Kafka cluster using the {@link GroupProtocol#CLASSIC classic group protocol}.
 * In this implementation, all network I/O happens in the thread of the application making the call.

经典实现里，网络 I/O 发生在调用 poll 的应用线程上。你调用 poll()，它才去发 fetch、收响应、推进位移。另有一个心跳线程负责维持组成员身份，但数据面全在应用线程。

AsyncKafkaConsumer.java 第 172–174 行：

 * <em>Note:</em> this {@link Consumer} implementation is part of the revised consumer group protocol from KIP-848.
 * This class should not be invoked directly; users should instead create a {@link KafkaConsumer} as before.
 * This consumer implements the new consumer group protocol and is intended to be the default in coming releases.

新实现把网络 I/O 移到了独立的后台线程（ConsumerNetworkThread），应用线程通过事件队列与它交互。poll() 不再直接发请求，而是提交一个 AsyncPollEvent，然后等待后台线程把数据准备好。

这个差异会直接体现在调优上：

维度ClassicKafkaConsumerAsyncKafkaConsumer
网络 I/O 线程调用 poll 的应用线程独立后台线程
心跳独立心跳线程后台线程统一管理
分配决策客户端 assignor 计算服务端计算（KIP-848）
可控的分配策略partition.assignment.strategy由 broker 的 group.consumer.assignors 决定
迁移定位向后兼容（pre-KIP-848）新协议目标实现

注意最后一行：官方对经典实现的定位是「exists for backward compatibility」，对新实现的定位是「intended to be the default in coming releases」。方向是清楚的，但 4.3.0 的默认值仍是经典实现。

为什么现在还有两套

一句话：迁移需要回滚路径。

KIP-848 改的是消费组协议本身——谁来做分配、成员状态怎么维护、再平衡怎么触发。这种改动不可能让所有存量应用一次性切换。所以客户端同时携带两套实现，用 group.protocol 做开关：

  • 存量应用不配这个参数，行为与升级前一致；
  • 想试新协议的应用显式写 group.protocol=consumer；
  • 出问题改回 classic 即可，不需要降级客户端版本。

这个设计对使用者的要求是：不要在代码里假设实现。所有依赖 ClassicKafkaConsumer 或 AsyncKafkaConsumer 具体类型的代码，都会在切换配置时失效。

选择时要注意的几点

第一，显式写出 group.protocol。 依赖默认值意味着升级客户端后行为可能变化。把 classic 或 consumer 写进配置，至少让变更可见。

第二，不要 instanceof 或强转。 KafkaConsumer 的运行时类型永远是门面类，强转到内部实现会直接 ClassCastException。即使通过反射拿到 delegate，也不要在业务代码里用——内部接口没有兼容性承诺。

第三，两套实现不要在同一个 group.id 下混跑。 这一点本机没有做完整验证，所以只给结论不给数据：迁移期应该整组切换，而不是逐个实例切换。混合状态下成员对协议的理解不一致，行为不可预期。真要灰度，用不同的 group.id 分流，而不是同一个组里一半经典一半新协议。

第四，注意配置黑名单。 如果你的通用配置模板里带了 session.timeout.ms 或 partition.assignment.strategy，切到 consumer 协议会在构造期直接失败。上面那条 ConfigException 就是这种情况。

第五，别把「默认 classic」当成长期结论。 官方注释明确说新实现将来会成为默认。现在写的代码，应该按「随时可能切换」来设计。

一个容易忽略的细节

ConsumerDelegateCreator 有两个 create 重载。第一个是我们看的公开路径，只接收 ConsumerConfig 和两个反序列化器；第二个接收 LogContext、Time、KafkaClient、SubscriptionState 等一整套依赖，供测试和内部构造使用。

两个重载的分支逻辑完全一样，都是 GroupProtocol.CONSUMER 走 Async，否则走 Classic。区别只在依赖注入方式。读源码时不要只看到一个就说「只有一条构造路径」。

另外，AsyncKafkaConsumer 的构造签名比 ClassicKafkaConsumer 多一个 Optional.empty() 参数。这是给内部测试注入用的，不影响公开 API。

本节判据

读完这一节，你应该能回答四个问题：

  1. new KafkaConsumer<>(props) 返回的是什么类？ 答：org.apache.kafka.clients.consumer.KafkaConsumer，一个门面；真实实现在私有 delegate 字段里。
  2. 默认走哪套？ 答：classic，因为 DEFAULT_GROUP_PROTOCOL 是 GroupProtocol.CLASSIC.name().toLowerCase(...)。
  3. 判定逻辑写在哪？ 答：ConsumerDelegateCreator.create，条件是 groupProtocol == GroupProtocol.CONSUMER。
  4. 怎么确认运行时实现？ 答：getClass() 不行；可以看构造期配置校验、日志类名或行为差异，不要依赖反射私有字段。

这四个问题的答案，决定了你在排查「为什么换了配置行为就变了」时，会不会往错误的方向找。

下一节进入数据面：不管哪套实现，记录最终都要经过一条拉取管线。Fetcher 在 4.3.0 只剩 9.1 KB，它的职责被拆到了 AbstractFetch、FetchCollector、CompletedFetch 里。我们会把 poll 的一次完整路径走一遍。

阅读导航：上一节 7.3 Sender 与 NetworkClient ｜ 下一节 8.2 Fetcher 与拉取管线

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收