使用 Kafka Binder 进行分区
Apache Kafka 原生支持主题分区。
有时将数据发送到特定分区是有利的 —— 例如，当您希望严格排序消息处理时（特定客户的所有消息应发送到同一分区）。
以下示例展示了如何配置生产者和消费者端：
@SpringBootApplication
public class KafkaPartitionProducerApplication {
private static final Random RANDOM = new Random(System.currentTimeMillis());
private static final String[] data = new String[] {
"foo1", "bar1", "qux1",
"foo2", "bar2", "qux2",
"foo3", "bar3", "qux3",
"foo4", "bar4", "qux4",
};
public static void main(String[] args) {
new SpringApplicationBuilder(KafkaPartitionProducerApplication.class)
.web(false)
.run(args);
}
@Bean
public Supplier<Message<?>> generate() {
return () -> {
String value = data[RANDOM.nextInt(data.length)];
System.out.println("Sending: " + value);
return MessageBuilder.withPayload(value)
.setHeader("partitionKey", value)
.build();
};
}
}
spring:
cloud:
stream:
bindings:
generate-out-0:
destination: partitioned.topic
producer:
partition-key-expression: headers['partitionKey']
partition-count: 12
需要注意的是，由于 Apache Kafka 本身支持分区，因此除非你像示例中那样使用自定义分区键或涉及有效负载本身的表达式，否则不需要依赖上述的绑定器分区功能。绑定器提供的分区选择功能主要是为那些不支持原生分区的中间件技术设计的。请注意，在上面的示例中，我们使用了一个名为 partitionKey 的自定义键，这将是分区的决定因素，因此在这种情况下，使用绑定器分区是合适的。当使用 Kafka 原生分区时，即当你没有提供 partition-key-expression 时，Apache Kafka 将选择一个分区，默认情况下，该分区将是记录键的哈希值除以可用分区数的余数。要为出站记录添加键，可以在 spring-messaging 的 Message<?> 中将 KafkaHeaders.KEY 头设置为所需的键值。默认情况下，当未提供记录键时，Apache Kafka 将根据 Apache Kafka 文档 中描述的逻辑选择分区。
主题必须配置足够的分区，以实现所有消费者组所需的并发性。上述配置最多支持 12 个消费者实例（如果它们的 concurrency 为 2，则支持 6 个；如果它们的并发性为 3，则支持 4 个，依此类推）。通常最好“超额配置”分区，以便将来增加消费者或并发性。
前面的配置使用了默认的分区策略（key.hashCode() % partitionCount）。这可能提供也可能不提供合适的平衡算法，具体取决于键值。特别要注意的是，这种分区策略与独立 Kafka 生产者（例如 Kafka Streams 使用的生产者）的默认策略不同，这意味着相同的键值在这些客户端生成时可能会在不同分区之间以不同的方式平衡。您可以通过使用 partitionSelectorExpression 或 partitionSelectorClass 属性来覆盖此默认行为。
由于分区由 Kafka 原生处理，因此在消费者端不需要特殊配置。Kafka 会在实例之间分配分区。
Kafka 主题的 partitionCount 在运行时可能会发生变化（例如由于管理任务）。之后计算的分区将会不同（例如将使用新的分区）。从 Spring Cloud Stream 4.0.3 开始，运行时分区数量的变化将得到支持。另请参阅参数 spring.kafka.producer.properties.metadata.max.age.ms 来配置更新间隔。由于一些限制，无法使用引用消息 payload 的 partition-key-expression，在这种情况下该机制将被禁用。默认情况下，整体行为是禁用的，可以通过配置参数 producer.dynamicPartitionUpdatesEnabled=true 来启用。
以下 Spring Boot 应用程序监听 Kafka 流，并将每条消息发送到的分区 ID 打印到控制台：
@SpringBootApplication
public class KafkaPartitionConsumerApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(KafkaPartitionConsumerApplication.class)
.web(WebApplicationType.NONE)
.run(args);
}
@Bean
public Consumer<Message<String>> listen() {
return message -> {
int partition = (int) message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION);
System.out.println(message + " received from partition " + partition);
};
}
}
spring:
cloud:
stream:
bindings:
listen-in-0:
destination: partitioned.topic
group: myGroup
你可以根据需要添加实例。Kafka 会重新平衡分区分配。如果实例数量（或 实例数量 * 并发数）超过分区数量，某些消费者将处于空闲状态。