Consumer 基础入门

在 Kafka Clients 模块中不仅提供了前面介绍的 KafkaProducer，还提供了 KafkaConsumer 的实现，核心实现位于 org.apache.kafka.clients.consumer 包中。

废话不多说了，先来看一下 KafkaConsumer 的基础使用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

public class ConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        // kafka集群地址       
        props.put("bootstrap.servers", "localhost:9092");
        // 所属Consumer Group的Id
        props.put("group.id", "consumer-demo");
        // 自动提交offset
        props.put("enable.auto.commit", "true");
        // 自动提交offset的时间间隔
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        // key和value使用的Deserializer
        props.put("key.deserializer", 
                  "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", 
                  "org.apache.kafka.common.serialization.StringDeserializer");
        // 根据上面的配置创建KafkaConsumer对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 同时订阅test_topic1和test_topic2
        consumer.subscribe(Arrays.asList("test_topic1", "test_topic2"));
        try {
            while (true) {
                // 从kafka集群拉取message，每次poll()可以拉取多个消息
                ConsumerRecords<String, String> records = consumer.poll(100);
                // 消费message，这里将消息的offset、key、value输出
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s\n", 
                                     record.offset(), record.key(), record.value());
            }
        } finally {
            // 关闭Consumer
            consumer.close();
        }
    }
}

我们可以通过前文介绍的 ProducerDemo 发送 message，然后使用上面的 ConsumerDemo 来消费 message，这里就不再展示效果了。

Consumer 接口

了解了 KafkaConsumer 的基本使用之后，我们来看 KafkaConsumer 的具体实现，首先是其实现的接口 —— Consumer，对 Consumer 接口中方法进行分类之后，我们可以得到如下三大类：

• 拉取元数据（查 topic、partition、offset）：
  • listTopics() 方法：查询 kafka 集群中全部 topic 的元信息。
  • partitionsFor() 方法：查询指定 topic 的元数据。
  • groupMetadata() 方法：查询当前 consumer group 的元信息。
  • beginningOffsets() 方法、endOffsets() 方法：查询指定 partition 的最小 offset 值（因为日志压缩和日志保留策略的存在，所以最小 offset 不一定是 0）以及最大 offset 值。
  • offsetsForTimes() 方法：根据时间戳查询指定 partition 中的 offset 值，返回 offset 值是大于指定时间的最小 offset（earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition）。
• 控制消费行为：
  • assign(Collection<TopicPartition>) 方法：用来手动给 consumer 分配其订阅的 topic 以及 partition 集合，该方法与 subscribe() 方法冲突。
  • subscribe()方法：订阅指定的 topic，后续将由 kafka 集群通过 rebalance 协议为 consumer 自动分配 partition。
  • commit()、commitAsync() 方法：提交（异步提交）当前 consumer 的消费位置（也就是其已消费完成的 message 的 offset 值），默认情况下，consumer 的消费位置存储在 __consumer_offsets 这个特殊的 topic 中。
  • poll() 方法：从当前 consumer 负责消费的 partition 中批量拉取 message。
  • seek(TopicPartition)方法：指定 consumer 从指定 partition 的哪个 offset 位置开始后续消费。
  • pause()、resume()方法：暂停/继续 consumer，暂停后 poll() 方法会返回集合。
• 获取消费装填：
  • assignment()：查询当前 consumer 消费的 partition。
  • position(TopicPartition) 方法：获取 consumer 下次从指定 partition 拉取的 message 的 offset 值。
  • committed() 方法：当前 consumer group 在指定 partition 的消费位置 。

ConsumerNetworkClient

在前面介绍 KafkaProducer 的时候详细介绍了 NetworkClient 的实现，它负责管理 Producer 与 kafka 集群中各个 broker 之间的网络连接，底层依赖 KSelector 来完成网络读写操作，通过一系列 handle*() 方法处理正常响应、超时以及网络异常等。

ConsumerNetworkClient 是专门为 Consumer 端提供的客户端，它底层依赖 NetworkClient 的上述功能来完成网络操作。依旧本着是先看数据结构，再看行为的思路，来看 ConsumerNetworkClient 的核心字段，如下：

send

作为一个 Client，ConsumerNetworkClient 自然提供了发送请求的方法 —— send() 方法，它会将待发送的请求封装成ClientRequest 对象，保存到unsent集合中等待发送，具体实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public RequestFuture<ClientResponse> send(Node node, AbstractRequest.Builder<?> requestBuilder,
				int requestTimeoutMs) {
    long now = time.milliseconds();
    // RequestFutureCompletionHandler是ConsumerNetworkClient中定义的一个内部类，实现了RequestCompletionHandler接口
    RequestFutureCompletionHandler completionHandler = new RequestFutureCompletionHandler();
    // 创建ClientRequest对象
    ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, 
					now, true, requestTimeoutMs, completionHandler);
    // 将ClientRequest对象写入到unsent中相应的队列中
    unsent.put(node, clientRequest);
    // 唤醒在NetworkClient.poll()方法上阻塞等待网络事件的线程
    client.wakeup();
    // 返回RequestFuture
    return completionHandler.future;
}

我们可以看到，这里创建 ClientRequest 时指定的 callback 对象是 RequestFutureCompletionHandler，其中关联了发送请求对应的 RequestFuture 对象（future 字段）， ClientResponse 对象（response 字段）以及相关的 RuntimeException（e 字段）。在其 onComplete() 方法中，会初始化 ClientResponse 对象并将当前 RequestFutureCompletionHandler 对象添加到 pendingCompletion 集合；onFailure() 方法中会初始化 Exception 并将当前 RequestFutureCompletionHandler 对象添加到 pendingCompletion 集合。

RequestFutureCompletionHandler 中的 future 字段是 RequestFuture 类型， RequestFuture 是 ConsumerNetworkClient 发出的一个异步请求的 Future 对象，其中的 result 字段（AtomicReference<Object> 类型）用来记录请求结果（具体 result 类型由 RequestFuture 的泛型指定），completedLatch 字段（CountDownLatch）用来实现 RequestFuture 的 awaitDone() 方法的阻塞效果，listeners 集合记录了（ConcurrentLinkedQueue）用来记录相关监听器。

在 RequestFutureCompletionHandler.fireCompletion() 方法中会根据 e 字段调用 RequestFuture 的相应方法，完成（或异常完成）该 RequestFuture，关闭 completedLatch，唤醒阻塞在其上的线程。

RequestFuture 中的方法都比较简单，熟悉 JDK Future 的胖友应该都能看懂。这里说一下 RequestFuture 中两个有意思的方法，一个是 compose(RequestFutureAdapter) 方法，它主要是将当前 RequestFuture 对象转换成另一个 RequestFuture 对象，目的是为了转换 result 类型，具体实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

public <S> RequestFuture<S> compose(final RequestFutureAdapter<T, S> adapter) {
    // 创建一个新的RequestFuture对象
    final RequestFuture<S> adapted = new RequestFuture<>();
    // 添加一个监听器，当前RequestFuture完成的时候，会通过该监听器触发新的RequestFuture完成
    addListener(new RequestFutureListener<T>() {
        @Override
        public void onSuccess(T value) {
            adapter.onSuccess(value, adapted);
        }

        @Override
        public void onFailure(RuntimeException e) {
            adapter.onFailure(e, adapted);
        }
    });
    return adapted; // 返回新的RequestFuture
}

另一处是 chain() 方法，它是为了将多个有依赖关系的 RequestFuture 组成一个责任链，具体实现原理与 compose() 方法类似，也是通过监听器的方式实现的。

poll

ConsumerNetworkClient 中最核心的方法是 poll(Timer timer, PollCondition pollCondition, boolean disableWakeup) 方法，这三个参数的含义如下：

• timer（Timer 类型）：Timer 是一个辅助类，用来帮助 NetworkClient.poll() 方法这种阻塞的方法计算超时时间。

• pollCondition（PollCondition 类型）：PollCondition 类似于一个钩子方法，通过外部提供的 shouldBlock() 钩子方法来控制 ConsumerNetworkClient.poll() 是否该进行阻塞。这里说一下 PollCondition 接口，它只有两个实现：

  

  RequestFuture 中的 shouldBlock() 方法实际就是 !isDone() 方法，也就是说，我们在 ConsumerNetworkClient.poll() 方法中根据某个请求的状态，决定是否阻塞等待其响应。

• disableWakeup（boolean 类型）：标识此次 poll() 方法是否是可被唤醒的，也就是，是否关注 wakeup 字段的值。如果 disableWakeup 字段设置为 true，则在 poll() 方法中不会关注 wakeup 字段值；如果设置为 false，则 wakeup 为 true，则抛出 WakeupException 并退出此次 poll() 方法调用。

好了，下面来看 poll() 方法的核心逻辑：

1、 加锁，ConsumerNetworkClient 是线程安全，毕竟有很多线程会并发使用 ConsumerNetworkClient 这个对象来发送请求并接收响应。

2、 执行 handlePendingDisconnects() 方法处理 pendingDisconnects 集合，此处会清理 UnsentRequests 中每个 Node 对应的 ClientRequest 集合，并触发每个 ClientRequest 关联的 RequestFutureCompletionHandler 的 onFailure() 方法。同时，还会调用 NetworkClient 的 disconnect() 方法，清理 InFlightRequests 集合并触发相关回调。

3、 执行 trySend() 方法，尝试发送 unsent 集合中的请求。具体逻辑是：遍历每个 Node 对应的 ClientRequest 列表，每次循环都调用 NetworkClient.ready() 方法检测当前是否具备发送条件，其中包括网络连接状态、InFlightRequests 队列状态等等。检测通过之后，则调用 NetworkClient.send() 方法将请求放入InFlightRequests 队里中等待响应，也放入 KafkaChannel 的 send 字段中等待发送。同时，会将请求从 unsent 中删除。

4、 接下来根据条件决定 NetworkClient.poll() 方法的阻塞时长，这里的条件有两个：

• pendingCompletion 集合是否为空，如果不为空，则表示有 RequestFutureCompletionHandler 等待执行，所以阻塞时长为 0。
• pollCondition.shouldBlock() 这个钩子方法的返回值，如果返回 false，则表示无需阻塞。其实就是看自己关注的请求是否已经完成了。

5、 在 NetworkClient.poll() 方法中会更新连接状态，在 poll() 方法结束之后，这里会再次检查连接状态并更新处理 unsent 集合中未发送的请求，具体逻辑与步骤1基本一致，不再重复。

6、 检查 disableWakeup、wakeup，决定是否抛出 WakeupException 异常。检查当前线程是否被中断，如果被中断了，则抛出 InterruptException 异常。

7、 再次调用 trySend() 方法，之所以再次调用 trySend() 方法，是因为经历了一次 poll() 之后，可能有的连接、InFlightRequests 等条件准备好了。

8、 调用 failExpiredRequests() 方法处理 unsent 集合中过期的请求，这里会调用过期请求关联的 RequestFutureCompletionHandler.onFailure() 方法。

9、 清空 unsent 集合。

10、 执行 firePendingCompletedRequests() 方法，触发 pendingCompletion 集合中全部的 RequestFutureCompletionHandler，这里不仅仅会完成 RequestFuture，还会触发注册在其上的监听器。

其他方法简介

• awaitMetadataUpdate()方法：循环调用 poll() 方法，直到 Metadata 更新完毕（版本号增加），实现阻塞等待Metadata 更新完成。
• awaitPendingRequests()方法：阻塞等待 unsent 和 InFightRequests 中的请求全部完成（正常完成或异常完成都可以）。
• leastLoadedNode() 方法：查找 kafka 集群中负载最低的Node。
• hasReadyNodes() 方法：查询当前 kafka 集群是否有 Node 可以发送请求。
• disconnectAsync() 方法：关闭与指定 Node 的连接，这里会写入 pendingDisconnects 集合，并调用 NetworkClient 的 wakeup() 方法，尽快处理断开的连接。

SubscriptionState

通过前面的介绍我们知道，在 consumer 启动之后会经历 rebalance 来确定自己消费的 partition，紧接着 consumer 会从 __consumer_offsets 这个内部 topic 拉取 offset 确认自己从哪个位置开始消费。

KafkaConsumer 在内存中使用 SubscriptionState 来维护自身消费的 topic、partition 以及 offset。下面来看 SubscriptionState 的核心字段：

这里我们来关注一下 assignment 字段，TopicPartitionState 用来记录当前 consumer 消费对应 partition 的消费状态，核心字段如下：

TopicPartitionState 提供了非常多的方法来读写上述字段，这里不再展开，没啥难度。

接下来看 PartitionStates，其中最核心的就是 map 字段（LinkedHashMap 类型），其中的 Key 是 TopicPartition，Value 是 TopicPartitionState，这里之所以使用 LinkedHashMap 类型的 Map，是为了更新 TopicPartition 的顺序，保证 Fetch Request 不会出现饥饿的状态，胖友们可以看一下 updateAndMoveToEnd() 、moveToEnd() 方法，都是用来调整 TopicPartition 消费状态顺序的。另外，PartitionStates 中提供了一些其他的方法都是用来读写 map 集合，不再多说。

到此为止，SubscriptionState 的核心数据结构就介绍完了。SubscriptionState 本身提供的行为基本都是用来修改上述字段的，这里不再展开了。

ConsumerCoordinator

在 KafkaConsumer 中通过 ConsumerCoordinator 实现与服务端的 GroupCoordinator 的完成 rebalance 协议中的一系列交互，ConsumerCoordinator 继承了 AbstractCoordinator 抽象类。AbstractCoordinator 的核心字段如下：

ConsumerPartitionAssignor

在 rebalance 协议里面，group leader 在收到 JoinGroup Response 之后，会按照其中指定的分配策略完成 partition 分配，每个 partition 分配策略就是一个 ConsumerPartitionAssignor 接口的实现。下图展示了 ConsumerPartitionAssignor 继承结构：

ConsumerPartitionAssignor 接口中定义了五个内部类，我们一个个来说：

• Subscription 中保存了 consumer 参加 rebalance 之前已经订阅的 topic（topics字段）、已分配的 partition（ownedPartitions 字段，List<TopicPartition>），以及在 consumer 端配置的 groupInstanceId（在 Static Membership 优化中使用）。

• Assignment 中保存了 partition 的分配结果，partitions 字段（List<TopicPartition>）表示的是分配给某 consumer 的 TopicPartition 集合，userData 是用户自定义的数据。

• GroupAssignment 底层通过 Map<String, Assignment> 维护全部的 partition 分配结果。

• GroupSubscription 底层通过 Map<String, Subscription> 记录全部 consumer 目前已分配的 partition 信息。这两个 Map 的 Key 都是 consumer 在 consumer group 中的唯一标识 —— MemberId。

• RebalanceProtocol 是个枚举，可选值有 EAGER、COOPERATIVE 两个，对应了我们前面介绍的两个 rebalance 协议的名称。

接下来看 ConsumerPartitionAssignor 的方法：

• assign() 方法是其中最核心的方法 ，它接收集 kafka 集群元数据信息以及每个 consumer 目前的 partition 订阅情况，返回的是 GroupAssignment，也就是 partition 的分配结果。
• 在每个 consumer 收到 group leader 发来的 partition 分配结果时，会调用 onAssignment() 这个回调方法，此调用发生在解析 SyncGroupResponse 之后。
• supportedProtocols() 方法用于获取当前 ConsumerPartitionAssignor 支持的 rebalance 协议。

AbstractPartitionAssignor 抽象类实现了 ConsumerPartitionAssignor 接口，其中主要是对 assign() 方法进行了简单的实现，具体实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription) {
    // 获取当前各个consumer已订阅的partition信息
    Map<String, Subscription> subscriptions = groupSubscription.groupSubscription();
    // 获取订阅了哪些topic
    Set<String> allSubscribedTopics = new HashSet<>();
    for (Map.Entry<String, Subscription> subscriptionEntry : subscriptions.entrySet())
        allSubscribedTopics.addAll(subscriptionEntry.getValue().topics());
    // 从Metadata元数据中获取每个topic的partition个数
    Map<String, Integer> partitionsPerTopic = new HashMap<>();
    for (String topic : allSubscribedTopics) {
        Integer numPartitions = metadata.partitionCountForTopic(topic);
        if (numPartitions != null && numPartitions > 0)
            partitionsPerTopic.put(topic, numPartitions);
    }
    // 调用子类的assign()方法进行partition分配
    Map<String, List<TopicPartition>> rawAssignments = assign(partitionsPerTopic, subscriptions);

    // 将partition分配结果封装成GroupAssignment对象返回
    Map<String, Assignment> assignments = new HashMap<>();
    for (Map.Entry<String, List<TopicPartition>> assignmentEntry : rawAssignments.entrySet())
        assignments.put(assignmentEntry.getKey(), new Assignment(assignmentEntry.getValue()));
    return new GroupAssignment(assignments);
}

RangeAssignor

下面来看 ConsumerPartitionAssignor 具体的实现类，RangeAssignor 策略的原理是按照 consumer 总数和 partition 总数进行整除运算来获得一个跨度，然后将 partition 按照跨度进行平均分配，以保证 partition 尽可能均匀地分配给所有的 consumer。

对于每一个 topic，RangeAssignor 策略会将 consumer group 内所有订阅这个 topic 的 consumer 按照名称的字典序排序，然后为每个 consumer 划分固定的 partition 范围，如果不够平均分配，那么字典序靠前的 consumer 会被多分配一个分区。

假设 n = partition_num / consumer_num，m = partition_num % consumer_num，那么前 m 个消费者每个分配 n+1 个 partition，后面的 consumer_num - m 个消费者每个分配 n 个 partition。我们举个栗子，假设 consumer group 内有 2 个 consumer，都订阅了 t0 和 t1 两个 topic，并且每个 topic 都有 4 个 partition，那么通过 RangeAssignor 分配的结果如下：

1
2

consumer0：t0p0、t0p1、t1p0、t1p1（其中，t0p0 是 topic0 中的 partition0）
consumer1：t0p2、t0p3、t1p2、t1p3

看上去 RangeAssignor 的分配是相对均匀的，但是在有的场景中，分配结果就会出现不均匀的情况。假设上面的示例中，每个 topic 有三个 partition，RangeAssignor 的分配结果如下：

1
2

consumer0：t0p0、t0p1、t1p0、t1p1
consumer1：t0p2、t1p2

明显可以看到，这种分配结果显然是不均匀的，随着不均匀的 topic 越来越多，会导致 consumer 处理的 partition 不均衡。

RoundRobinAssignor

RoundRobinAssignor 分配策略的原理是将 consumer group 内所有 consumer 以及所有 TopicPartition 按照字典序排序，然后通过轮询方式逐个将 partition 分配给每个 consumer。

如果同一个 consumer group 内所有 consumer 订阅的 topic 都是相同的， RoundRobinAssignor 策略的分配结果是均匀的。依旧是上面两个 t0、t1 两个 topic，每个 topic 三个 partition，RoundRobinAssignor 的分配结果为：

1
2

consumer0：t0p0、t0p2、t1p1
consumer1：t0p1、t1p0、t1p2

但如果 consumer group 内的 consumer 订阅了不同的 topic，有可能会导致分配结果不均匀。例如，假设 consumer group 内有 3 个 consumer，订阅了 3 个 topic，每个 topic 有 3 个 partition。其中，consumer0 订阅了 topic0，consumer1 订阅了 t0 和 t1 ，consumer2 订阅的是 t0、t1 和 t2，最终的分配结果如下：

1
2
3

consumer0：t0p0
consumer1：t1p0
consumer2：t1p1、t2p0、t2p1、t2p2

StickyAssignor

StickyAssignor 是一种黏性分配策略，它的主要目的是在分配均匀的前提下，尽可能的与上次分配结果保持一致。StickyAssignor策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂很多。

下面我们通过一个示例来说明 StickyAssignor 的分配效果。假设 consumer group 内有 3 个 consumer，他们同时订阅了 4 个topic，每个 topic 有 2 个 partition。最终的分配结果如下：

1
2
3

consumer0：t0p0、t1p1、t3p0
consumer1：t0p1、t2p0、t3p1
consumer2：t1p0、t2p1

此时，consumer1 如果宕机离开 consumer group，那么 consumer group 会执行rebalance，新的分配结果就变成了：

1
2

consumer0：t0p0、t1p1、t3p0、t2p0
consumer2：t1p0、t2p1、t0p1、t3p1

而如果使用 RoundRobinAssignor ，分配的结果是：

1
2

consumer0：t0p0、t1p0、t2p0、t3p0
consumer2：t0p1、t1p1、t2p1、t3p1

比较这两个策略的分配结果可以看出，StickyAssignor 保留了上一次分配的全部结果，只是将 consumer1 负责的 partition 分配给了 consumer0 和 consumer1。

再举个栗子，三个 consumer订阅了三个 topic，每个 topic 三个 partition，consumer0 订阅了 t0，consumer1 订阅了 t0 和 t1，consumer2 订阅了 t0、t1 和 t2。此时采用 RoundRobinAssignor 策略得到的不均匀结果前面说明了，这里不再重复。如果使用 StickyAssignor 策略，分配结果为：

1
2
3

consumer0：t0p0
consumer1：t1p0, t1p1
consumer2：t2p0, t2p1, t2p2

如果此时 consumer0 宕机，RoundRobinAssignor策略产生的新分配结果为：

1
2

consumer1：t0p0、t1p1
consumer2：t1p0、t2p0、t2p1、t2p2

而使用 StickyAssignor 策略产生的新分配结果为：

1
2

consumer1：t1p0、t1p1、t0p0
consumer2：t2p0、t2p1、t2p2

显然，StickyAssignor 策略比其他两种分配策略更加优秀。

下面来看一下 StickyAssignor 的实现吧，为啥不看 RangeAssignor、RoundRobinAssignor 的实现呢？因为简单，简单到从原理直接就可以给出代码。

具体实现分析

要看 StickyAssignor 先父类 AbstractStickyAssignor，它的 assign() 方法实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13

public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    Map<String, List<TopicPartition>> consumerToOwnedPartitions = new HashMap<>();
    if (allSubscriptionsEqual(partitionsPerTopic.keySet(), subscriptions, consumerToOwnedPartitions)) {
        log.debug("Detected that all consumers were subscribed to same set of topics, invoking the optimized assignment algorithm");
        partitionsTransferringOwnership = new HashMap<>();
        return constrainedAssign(partitionsPerTopic, consumerToOwnedPartitions);
    } else {
        log.debug("Detected that all not consumers were subscribed to same set of topics, falling back to the general case assignment algorithm");
        partitionsTransferringOwnership = null;
        return generalAssign(partitionsPerTopic, subscriptions);
    }
}

allSubscriptionsEqual() 方法中做了两件事，一件事是检查各个 consumer 订阅的 topic 是否一致，体现在该方法的返回值中，另一件事是填充 consumerToOwnedPartitions 集合，这个集合填充的是当前各个 consumer 处理的 TopicPartition 集合。深入 allSubscriptionsEqual() 方法可以看到填充 consumerToOwnedPartitions 集合的数据来源与 memberData() 这个抽象方法，StrickyAssignor 的 实现是从 Subscription.userData 中得到的。

如果 consumer group 中的 consumer 都订阅了同一个 topic，会进入 constrainedAssign() 方法这个分支，走优化算法。constrainedAssign() 方法的核心步骤是：

1、 执行一些准备工作，例如：

• 计算每个 consumer 能够分配的最大 partition 数（maxQuota）和最小 partition 数（minQuota）
• 用来存储分配 partition 已达上限的 consumer 的集合（maxCapacityMembers 集合）
• 用来存储分配 partition 已达下限的 consumer 的集合（minCapacityMembers 集合）
• 用来存储分配 partition 未达下限的 consumer 的集合（unfilledMembers 集合）。
• 存储全部未分配的 TopicPartition 的 unassignedPartitions 集合。
• 存储从任意 consumer revoke 回来的 TopicPartition 的 allRevokedPartitions 集合。
• 存储最终分配结果的 assignment 集合（Map<String, List<TopicPartition>>）。

2、 根据 consumerToOwnedPartitions 重新分配 partition，分配出去的 partition 将从 unassignedPartitions 中删除：

• 碰到 TopicPartition 数超过 maxQuota 的 consumer，多出来的 TopicPartition 进入 allRevokedPartitions 集合。
• 碰到 TopicPartition 数不足 minQuota 的 consumer，consumer 进入 unfilledMembers 集合。
• 碰到 TopicPartition 数等于 maxQuota 或 minQuota 的consumer，consumer 进入 maxCapacityMembers 或 minCapacityMembers 集合。

3、 如果 unfilledMembers、unassignedPartitions 两个集合不为空，则表示同时有未达标的 consumer 和未分配的 partition，开始将 unassignedPartitions 中的 TopicPartition 分配给 unfilledMembers 中的 consumer。

4、 如果 unfilledMembers 还不为空，说明给 maxCapacityMembers 集合中 consumer 分配 partition 分多了，就要开始从 maxCapacityMembers 的 consumer 那里偷 partition 了，但是要有个限度，不能把原来的 maxCapacityMembers 的 consumer 薅到 minQuota 以下。

5、 最后，来处理 unfilledMembers 为空，但是 unassignedPartitions 不为空的情况，这说明 minCapacityMembers 里面有太多 consumer 摸鱼，这里就要将 unassignedPartitions 中 partition 分配给 minCapacityMembers 中的摸鱼 consumer，提高他们的负载，但是也是有上限的，给他们分配再多，不能超过 maxQuato。

经过上面的这一通处理，assignment 集合就是最终的分配结果了。

如果 consumer group 中的 consumer 并没有订阅了同一个 topic，会进入 generalAssign() 方法这个分支。generalAssign() 方法涉及到的核心数据结构如下：

• consumer2AllPotentialPartitions（Map<String, List<TopicPartition>>）、partition2AllPotentialConsumers（Map<TopicPartition, List<String>>），因为 consumer group 中的 consumer 订阅的 topic 不同，所以需要记录每个 consumer 能分配哪些 TopicPartition（consumer2AllPotentialPartitions），每个 TopicPartition 能分配给哪些 consumer（partition2AllPotentialConsumers） 。
• partitionsPerTopic: 每个 topic 有多少过 partition。
• subscriptions：每个 consumer 订阅了哪些 topic。
• currentAssignment（Map<TopicPartition, ConsumerGenerationPair>）记录了当前每个 consumer 分配的 partition 集合（就是此轮 rebalance 开始之前的 partition 分配结果）。后面会基于这个集合，得出此轮 rebalance 的结果。
• currentPartitionConsumer：记录了当前每个 Partition 分配给了哪个 Consumer，其实就是这个集合的翻转 currentAssignment （即 consumer –> List<TopicPartition> 转成了 TopicPartition –> consumer）。
• sortedPartitions：TopicPartition 的有序集合，排序规则是按照 TopicPartition 能分配给多少 consumer 顺序排的。
• unassignedPartitions：记录还未分配出去的 TopicPartition。
• sortedCurrentSubscriptions：consumer 的有序集合，排序规则是按照目前已分配的 partition 数进行排序的。
• reassignablePartitions：可重新分配的 TopicPartition 集合，
• fixedPartitions：不可重新分配的 TopicPartition 集合。
• fixedAssignments：不可修改其分配结果的 consumer 集合。

generalAssign() 算法的核心步骤如下：

1、更新 currentAssignment 集合，这里会删除不存在的 consumer 和 TopicPartition。

2、分配 unassignedPartitions 集合中的 TopicPartition：这里会按照 sortedCurrentSubscriptions 集合的顺序分配 TopicPartition（当然，前提是 consumer 订阅了这个 TopicPartition）。这一轮操作下来之后，可能还有 TopicPartition 未分配出去，也可能有 consumer 没有分配任何 TopicPartition。

3、查询分配结果不能修改分配结果的 consumer，将这些 consumer 记录到 fixedAssignments 集合，同时将这些 consumer 从 currentAssignment 集合删除。（缩小问题范围）

4、接下来按照 sortedPartitions 集合的顺序，处理可以重新分配的 TopicPartition。如果移动一个 TopicPartition 可以提高整体的平衡性的话，就进行移动，直至达到 balance 的状态。

5、如果在步骤 4 中移动了 TopicPartition，但是没有提高平衡性（ balance score）的话，就需要根据黏性条件，决定是否回滚到原来的分配结果。

6、最后，将 fixedAssignments 中记录的 consumer 重新添加回 currentAssignment 集合。

在 generalAssign() 方法中判断一次分配结果是否平衡的结果有两个：

• consumer 分配的 TopicPartition 数量只差不超过 1。
• 不存在重新分配 TopicPartition 提高平衡性的可能。这里平衡性（balance score）的计算方式是：所有 consumer 分配的 TopicPartition 数量的差值，这个差值越小，分配结果越平衡。

StickyAssignor 的算法核心与 AbstractStickyAssignor 一致，StickyAssignor 实现了序列化和反序列化 MemebrData 的方法，以及 onAssignment() 方法，这些方法没有什么难度，这里不展开了。

CooperativeStickyAssignor

CooperativeStickyAssignor 与 StickyAssignor 一样，都继承了 AbstractStickyAssignor 抽象类，其 assign() 方法首先会调用 super.assign() 方法完成 partition 分配。在分配过程中，如果一个 TopicPartition 发生了迁移（也就是更换了 consumer），则将该 TopicPartition 记录到 partitionsTransferringOwnership 集合中。最后，调用 adjustAssignment() 方法从分配结果中删除 partitionsTransferringOwnership 集合，从而满足 Incremental Cooperative Rebalancing 协议的需求。

总结

本课时重点介绍了 kafka consumer 中 partition 分配策略的具体实现。下一课时开始介绍 KafkaConsumer 的核心实现。

consumer group rebalance 原理

在开始分析 KafkaConsumer 的具体实现之前，我们先来介绍一下 KafkaConsumer 涉及到的一些基础理论。在第一课时介绍 Consumer Group 时提到，对于同一个 Consumer Group 来说，同一个 Topic 的不同 partition 会分配给不同的 consumer 进行消费，那如何分配 partition，如何在有新 consumer 加入以及 consumer 宕机的时候重新分配 partition，就是我们说的 consumer group rebalance。

通过上一课时的介绍我们了解到，业务线程通过 KafkaProducer.send() 方法将 message 放入 RecordAccumulator 中进行能缓冲，并没有进行实际的网络 I/O 操作，真正的网络 I/O 操作是由 Sender 线程完成。

通过上一课时的介绍我们了解到，业务线程使用 KafkaProducer.send() 方法发送 message 的时候，会先将其写入RecordAccumulator 中进行缓冲，当 RecordAccumulator 中缓存的 message 达到一定阈值的时候，会由 IO 线程批量形成请求，发送到 kafka 集群。本课时我们就重点来看一下 RecordAccumulator 这个缓冲区的结构。

两数之和

给定一个整数数组 nums 和一个整数目标值 target，请你在该数组中找出和为目标值的那两个整数，并返回它们的数组下标。
你可以假设每种输入只会对应一个答案。但是，数组中同一个元素在答案里不能重复出现。
你可以按任意顺序返回答案。

示例 1：

1
2
3

输入：nums = [11,2,15,7], target = 9
输出：[1,3]
解释：因为 nums[1] + nums[3] == 9 ，返回 [1, 3] 。

示例 2：

1
2

输入：nums = [3,2,4], target = 6
输出：[1,2]

示例 3：

1
2

输入：nums = [3,3], target = 6
输出：[0,1]

提示：

1
2
3
4

2 <= nums.length <= 103
-109 <= nums[i] <= 109
-109 <= target <= 109
只会存在一个有效答案

解法一：

暴力循环。以示例1为例，我们拿到 nums 数组之后，需要为每个值循环一边 nums 数组，这样的话时间复杂度就是 O(n^2)。

1
2
3
4
5
6
7
8
9
10
11
12

public static int[] twoSum(int[] numbers, int target) {
    int[] ret = new int[2];
    for (int i = 0; i < numbers.length; i++) {
        for (int j = i + 1; j < numbers.length; j++) {
            if (numbers[i] + numbers[j] == target) {
                ret[0] = i + 1;
                ret[1] = j + 1;
            }
        }
    }
    return ret;
}

虽然这种方法可以得到解，但是在面试的时候，面试官一般会问：还有没有更优解，甚至直接给挂掉。

解法二：

使用哈希表的方式来解决，核心思想是用空间换时间。

我们在遍历的过程中，用一个 HashMap 存储数值与其对应的下标索引，然后通过 target - 当前值 的方式，得到补值然后查 HashMap 即可确定两数的下标值。

1
2
3
4
5
6
7
8
9
10

public int[] twoSum(int[] numbers, int target) {
    Map<Integer, Integer> map = new HashMap<Integer, Integer>();
    for (int i = 0; i < nums.length; ++i) {
        if (map.containsKey(target - nums[i])) { // 检查补值是否存在
            return new int[]{map.get(target - nums[i]), i};
        }
        map.put(nums[i], i); // 补值不存在，则放入当前值
    }
    return new int[0];
}

这种解法的时间复杂度就是 O(n)，但是空间复杂度就上升到了 O(n)。

变体1

如果上述问题多加一个条件：输入的数组是一个有序数组的话，你该如何优化算法呢？

我们可以采用双指针的方式来解这道题，我们使用 i、j 两个指针，指向数组的头和尾，不断计算 num[i] 和 num[j] 的和，当和大于 target 的时候，j 前移，当和小于 target 的时候，i 后移，直至找到 i 和 j。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

public int[] twoSum(int[] numbers, int target) {
    if (numbers == null || numbers.length == 0)
        return null;
    int i = 0;
    int j = numbers.length - 1;
    while (i < j) {
        int x = numbers[i] + numbers[j];
        if (x < target) {
            ++i;
        } else if (x > target) {
            j--;
        } else {
            return new int[]{i + 1, j + 1};
        }
    }
    return null;
}

三数之和

输入一个无序的数组，从这个数组中找出 3 个数之和为 0 的所有组合。
注意：要求输出不重复的解。

示例

1
2
3
4
5
6
7
8
9

输入：
[-1, 0, 1, 2, -1, -4],

输出：
[
  [-1, 0, 1],
  [-1, -1, 2]
]
// [-1，-1，2]、[2, -1, -1] 这 3 个解是重复的。

解法一：

暴力解法，三层 for 循环，这里就不多说了。

解法二：

我们可以先对数组进行排序，接下来遍历数组，对于遍历到的每个值，再使用双指针的方式，查找两个与其和为 0 的值，查找一组结果就记录下来，直到遍历完成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

public ArrayList<ArrayList<Integer>> threeSum(int[] num) {
    ArrayList<ArrayList<Integer>> result = new ArrayList<ArrayList<Integer>>();
    if (num.length < 3)
        return result;
    // sort array
    Arrays.sort(num);
    for (int i = 0; i < num.length - 2; i++) {
        if (i == 0 || num[i] > num[i - 1]) {
            int negate = -num[i];
            int start = i + 1;
            int end = num.length - 1;
            while (start < end) {
                if (num[start] + num[end] == negate) {
                    ArrayList<Integer> temp = new ArrayList<Integer>();
                    temp.add(num[i]);
                    temp.add(num[start]);
                    temp.add(num[end]);
                    result.add(temp);
                    start++;
                    end--;
                    //avoid duplicate solutions
                    while (start < end && num[end] == num[end + 1])
                        end--;
                    while (start < end && num[start] == num[start - 1])
                        start++;
                } else if (num[start] + num[end] < negate) {
                    start++;
                } else {
                    end--;
                }
            }
        }
    }
    return result;
}

本课时的文章和视频讲解，还会放到：

• 微信公众号：
  

• B 站搜索：杨四正

力扣习题课主要是讲解 LeetCode 上的算法题，可以帮助你应对面试中的一些算法题。

1、两数之和

深潜 kafka producer —— 核心架构

kafka 自定义了一套网络协议，我们可以使用任意语言来实现这套协议，实现向 kafka 集群 push message 以及从 kafka 集群 pull message 的效果。在 kafka 2.8.0 版本的源码中的 clients 模块就是官方默认提供的 Java 版本 producer、consumer 实现，我们本课时重点关注其中的 producer 部分实现。

kafka 2.8.0 源码环境搭建

安装 JDK

JDK 的安装非常简单，这里我们安装 JDK 8 即可（虽然 JDK 现在的新版本是 16，但是国内生产环境还有相当一大部分还是停留在 8 这个版本上）。首先到 JDK下载地址下载对应系统的 JDK 安装包即可，这里我使用的是 mac 系统，下载 dmg 文件即可。

下载完成之后，双击 dmg 文件，一直下一步即可安装完成。