Java kafka如何实现自定义分区类和拦截器

一、自定义分区

Kafka 提供了默认的分区策略，默认分区策略为DefaultPartitioner。当我们需要实现自定义分区策略时，需要继承Partitioner接口，并重写其中的方法。下面是实现自定义分区的步骤：

1. 实现Partitioner接口

public class CustomPartitioner implements Partitioner {

    /**
     * 实现分区方法
     *
     * @param topic 主题
     * @param key   键
     * @param keyBytes   键的字节数组
     * @param value 值
     * @param valueBytes 值的字节数组
     * @param cluster Kafka集群
     * @return 分区号
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义分区逻辑
        // ...
    }

    /**
     * 释放资源
     */
    @Override
    public void close() {
        // ...
    }

    /**
     * 配置方法
     *
     * @param configs 配置信息
     */
    @Override
    public void configure(Map<String, ?> configs) {
        // ...
    }
}

1. 配置自定义分区

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("partitioner.class", "com.example.CustomPartitioner");// 配置自定义分区
// ...

运行后，Kafka 将使用自定义的分区策略。

二、自定义拦截器

Kafka 提供了接口Interceptor，通过实现此接口，可以实现自定义拦截器。下面是实现自定义拦截器的步骤：

1. 实现Interceptor接口

public class CustomInterceptor implements Interceptor<String, String> {

    /**
     * 初始化
     *
     * @param configs 配置信息
     */
    @Override
    public void configure(Map<String, ?> configs) {

    }

    /**
     * 拦截方法
     *
     * @param record 记录
     * @return 拦截后的记录
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 拦截逻辑
        // ...
        return record;
    }

    /**
     * 释放资源
     */
    @Override
    public void close() {

    }
}

1. 配置自定义拦截器

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, CustomInterceptor.class.getName());// 配置自定义拦截器类
// ...

运行后，Kafka 将使用自定义的拦截器。

示例1：自定义分区类

public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        Integer partitionCount = cluster.partitionCountForTopic(topic);
        return Math.abs(key.hashCode() % partitionCount);
    }

    @Override
    public void close() {

    }

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

    }
}

public static void main(String[] args) throws InterruptedException {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("partitioner.class", "com.example.CustomPartitioner");// 配置自定义分区
    Producer<String, String> producer = new KafkaProducer<>(props);
    for (int i = 0; i < 1000; i++) {
        String key = Integer.toString(i % 5);
        String value = Integer.toString(i);
        producer.send(new ProducerRecord<>("test_topic", key, value));
        System.out.println("发送消息 - key: " + key + ", value: " + value);
    }
    producer.close();
}

此示例中，我们使用自定义分区实现了按照 key 的 hashcode 值与主题中的分区数取模的方式来决定所属分区的逻辑。运行结果：

发送消息 - key: 0, value: 0
发送消息 - key: 1, value: 1
发送消息 - key: 2, value: 2
发送消息 - key: 3, value: 3
发送消息 - key: 4, value: 4
发送消息 - key: 0, value: 5
...

可以看到，同一 key 值的消息永远会被发送到相同的分区。

示例2：自定义拦截器

public class CustomInterceptor implements Interceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifiedValue = "custom_prefix_" + record.value();
        ProducerRecord<String, String> newRecord = new ProducerRecord<>(record.topic(), record.partition(),
                record.key(), modifiedValue);
        return newRecord;
    }

    @Override
    public void close() {

    }

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

    }
}

public static void main(String[] args) throws InterruptedException {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, CustomInterceptor.class.getName());// 配置自定义拦截器
    Producer<String, String> producer = new KafkaProducer<>(props);
    for (int i = 0; i < 5; i++) {
        String key = Integer.toString(i % 5);
        String value = Integer.toString(i);
        producer.send(new ProducerRecord<>("test_topic", key, value));
        System.out.println("发送消息 - key: " + key + ", value: " + value);
        Thread.sleep(1000);
    }
    producer.close();
}

此示例中，我们实现了一个简单的自定义拦截器，通过在原始消息的值前加上一个自定义前缀"custom_prefix_"来改变所发送的消息，具体示例如下：

发送消息 - key: 0, value: 0
发送消息 - key: 1, value: 1
发送消息 - key: 2, value: 2
发送消息 - key: 3, value: 3
发送消息 - key: 4, value: 4

可以看到，拦截器成功拦截了经过 Producer 发送的消息，并对其进行了修改。