springboot 整合 kafka 分区策略

在 Spring Boot 中整合 Kafka 时，分区策略是一个重要的概念。Kafka 通过分区策略将消息分布到不同的分区中，从而实现负载均衡和并行处理。以下是一些常见的分区策略：

1. 轮询（RoundRobin）：这是 Kafka 默认的分区策略。消费者组中的每个消费者按顺序从分区列表中获取消息。这种方式简单易用，但可能导致某些消费者过载。

2. 基于键的哈希（Keyed Hash）：这种策略根据消息键的哈希值进行分区。相同键的消息将被发送到同一个分区。这种方式可以保证相同键的消息顺序一致，但可能导致负载分布不均。

3. 基于消息内容的哈希（Content Hash）：这种策略根据消息内容的哈希值进行分区。相同内容的消息将被发送到同一个分区。这种方式可以保证相同内容的消息顺序一致，但可能导致负载分布不均。

要在 Spring Boot 中整合 Kafka 并使用自定义分区策略，你需要执行以下步骤：

1. 添加 Kafka 依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

2. 配置 Kafka 生产者和消费者：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: my-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3. 创建自定义分区策略类：

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;

public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Cluster cluster) {
        // 实现自定义分区策略
        // 返回分区编号
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}

4. 在 Kafka 配置类中使用自定义分区策略：

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

现在，你已经成功整合了 Kafka 并使用了自定义分区策略。在发送消息时，Kafka 会使用你定义的 CustomPartitioner 类来决定将消息发送到哪个分区。