kafka自定义组件

kafka对于部分配置提供了统一接口，允许用户自定义接口实现，根据业务需求进行定制，如自定义分区器以及序列化反序列化

自定义分区器

kafka的默认分区策略可能不能很好的满足业务需求，可以需要根据kafka提供的API开发定制满足业务场景的分区策略，自定义一个分区器

public interface Partitioner extends Configurable, Closeable {
        // 分区，自定义所传入的分区
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

    
    public void close();

    default public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
    }
}

public interface Configurable {
        //获取配置信息和初始化数据时调用
    void configure(Map<String, ?> configs);

}

• 实现org.apache.kafka.clients.producer.Partitioner接口，重写该接口的partition方法，在该方法中实现分区分配的算法

• 实例化KafkaProducer的配置中指定partitioner.class为自定义的分区器

// 自定义分区器
public class CustomPartition implements Partitioner {

    private static final Integer PARTITIONS = 6;

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 如果key为null，采用默认分区
        if (key == null) {
            System.out.println("key为null,采用默认分区");
            return new DefaultPartitioner().partition(topic, key, keyBytes, value, valueBytes, cluster);
        } else {
            String code = String.valueOf(key);
            try {
                int partitionId = code.hashCode() % PARTITIONS;
                System.out.println("-----------" + code.hashCode() + "-----------------" + partitionId);
                return partitionId;
            } catch (Exception e) {
                System.out.println("分区出现异常，采用分区0" + e.getMessage());
                return 0;
            }
        }
    }

    @Override
    public void close() {

    }

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

    }
}



public class TestPartition {
    /**
     * 设置生产消息的总数
     */
    private static final int MSG_SIZE = 10;
    /**
     * 主题名称
     */
    private static final String TOPIC_NAME = "test-topic";

    private static KafkaProducer<String, String> producer;

    static {
        Properties properties = initConfig();
        producer = new KafkaProducer<String, String>(properties);
    }

    /**
     * 初始化kafka配置
     *
     * @return
     */
    private static Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 配置分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CustomPartition.class.getName());
        return properties;
    }

    public static void main(String[] args) {
        ProducerRecord<String, String> record = null;
        try {
            for (int i = 0; i < MSG_SIZE; i++) {
                record = new ProducerRecord<>(TOPIC_NAME,"分区消息key"+i,"分区消息体"+i);
                producer.send(record);
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

自定义序列化

需要实现org.apache.kafka.common.serialization.Serializer接口

自定义拦截器

对于 producer 而言，interceptor 使得用户在消息发送前以及 producer 回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer 允许用户指定多个 interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)。Intercetpor 的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor

public interface ProducerInterceptor<K, V> extends Configurable {
    // 该方法会封装进 KafkaProducer.send 方法中，即它运行在用户主线程中。Producer 确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的 topic 和分区，否则会影响目标分区的计算
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);

    // 该方法会在消息从 RecordAccumulator 成功发送到 Kafka Broker 之后，或者在发送过程中失败时调用。并且通常都是在 producer 回调逻辑触发之前。onAcknowledgement 运行在producer 的 IO 线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢 producer 的消息发送效率
    public void onAcknowledgement(RecordMetadata metadata, Exception exception);

    //关闭 interceptor，主要用于执行一些资源清理工作
    public void close();
}

public interface Configurable {

    //获取配置信息和初始化数据时调用
    void configure(Map<String, ?> configs);

}

执行顺序

拦截器->序列化器->分区器