5 分钟了解 kafka consumer 的 subscribe 和 assign 区别和应用场景

磊叔的技术博客

发布于 2025-06-09 19:40:59

3770

在 Kafka 中，消费者在订阅主题时，可以选择 subscribe 或 assign 两种方法。两者的主要区别在于 消费者的分区分配 和 使用场景。下面是它们的详细区别和使用示例：

1. `subscribe` 和 `assign` 的区别

• subscribe：
- • 适用于 消费者组 的场景。
- • 当消费者使用 subscribe 方法订阅一个或多个主题时，Kafka 会将消费者加入指定的消费者组，并由 协调者（Coordinator） 自动管理分区的分配。
- • 适合用于需要 自动负载均衡 的情况，Kafka 会自动分配分区，处理消费者组中的 重平衡（Rebalance）。
- • 重平衡时，Kafka 会动态调整分区分配，确保每个分区只被一个消费者消费。

• assign：
- • 适用于 手动管理分区 的场景。
- • 当消费者使用 assign 方法时，它直接指定要消费的分区列表，而不是订阅一个主题。这种方式不会使用消费者组，也不涉及分区的自动分配和重平衡。
- • 适合在对分区有精确控制的场景中使用，尤其是希望每个消费者负责固定分区时。
- • 由于没有消费者组的概念，assign 不会有负载均衡，也不会触发重平衡。

img

2. `subscribe` 和 `assign` 的使用场景

适用场景：

• subscribe 适用场景：
- • 需要自动分区分配，或者希望 Kafka 自动管理消费者组的重平衡。
- • 适用于大多数分布式消费场景，尤其是在处理流式数据时，消费者数量可能会变化，Kafka 可以自动调整分配。
• assign 适用场景：
- • 希望手动管理分区分配，避免因为重平衡带来的分区重新分配。
- • 适用于需要严格控制分区消费情况，确保每个消费者只消费特定的分区，不希望受到其他消费者的影响。
- • 在需要实现类似于消息队列（Queue）模型的场景，assign 允许开发者自行设计负载均衡策略。

3. `subscribe` 和 `assign` 的示例代码

使用 `subscribe`

import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅主题 "my-topic" 和 "another-topic"
consumer.subscribe(Arrays.asList("my-topic", "another-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis());
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

在这个例子中，消费者通过 subscribe 订阅了 my-topic 和 another-topic 主题。Kafka 会根据当前消费者组内的消费者数量自动分配分区，并在消费者数量变化时进行重平衡。

使用 `assign`

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import java.util.Properties;
import java.util.Collections;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 手动指定主题和分区，例如 "my-topic" 的第 0 分区
TopicPartition partition0 = new TopicPartition("my-topic", );
consumer.assign(Collections.singletonList(partition0));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis());
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

在此例中，消费者通过 assign 方法直接订阅了 my-topic 的第 0 分区。这样不会触发消费者组的重平衡，适合需要精确控制的消费场景。

总结

方法	适用场景	分配机制	是否支持重平衡
subscribe	自动负载均衡的分布式消费	由 Kafka 自动分配分区	是
assign	手动控制分区消费	消费者指定具体分区	否（无消费者组概念）