快捷導(dǎo)航

kafka自定義分區(qū)器使用詳解

更新時(shí)間：2025年11月19日 11:20:38 作者：princeAladdin

本文介紹了如何根據(jù)企業(yè)需求自定義Kafka分區(qū)器,只需實(shí)現(xiàn)Partitioner接口并重寫(xiě)partition()方法,示例中,包含"cuihaida"的數(shù)據(jù)發(fā)送到0號(hào)分區(qū),否則發(fā)送到1號(hào)分區(qū),在生產(chǎn)者配置中添加分區(qū)器參數(shù)即可使用

kafka自定義分區(qū)器

根據(jù)企業(yè)需求，自己重新實(shí)現(xiàn)分區(qū)器

只需要定義類實(shí)現(xiàn)Partitioner接口，然后重寫(xiě)partition()方法即可

假設(shè)現(xiàn)在有一個(gè)需求

發(fā)送過(guò)來(lái)的數(shù)據(jù)中如果包含cuihaida，就發(fā)往0號(hào)分區(qū)，不包含cuihaida，就發(fā)往1號(hào)分區(qū)

package com.example.kafkademo.producer;

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

import java.util.Map;

/**
 * 1. 實(shí)現(xiàn)接口Partitioner
 * 2. 實(shí)現(xiàn)3個(gè)方法:partition,close,configure
 * 3. 編寫(xiě)partition方法,返回分區(qū)號(hào)
 */
public class MyPartitioner implements Partitioner {

    /**
     * 重寫(xiě)這個(gè)方法
     * @param topic 主題
     * @param key 消息的key
     * @param keyBytes 消息的key序列化后的字節(jié)數(shù)組
     * @param value 消息的值
     * @param valueBytes 消息的值序列化后的字節(jié)數(shù)組
     * @param cluster 集群元數(shù)據(jù)可以查看分區(qū)信息
     * @return 信息對(duì)應(yīng)的分區(qū)
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 獲取消息
        String msgValue = value.toString();
        // 發(fā)送過(guò)來(lái)的數(shù)據(jù)中如果包含cuihaida，就發(fā)往0號(hào)分區(qū)，不包含cuihaida，就發(fā)往1號(hào)分區(qū)
        return msgValue.contains("cuihaida") ? 0 : 1;
    }

    @Override
    public void close() {

    }

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

    }
}

使用分區(qū)器的方法

在生產(chǎn)者的配置中添加分區(qū)器參數(shù)

package com.example.kafkademo.util;

import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

public class CommonUtils {
    /**
     * kafka生產(chǎn)者配置配置
     * @return 配置內(nèi)容
     */
    public static Properties buildKafkaProperties() {
        // 1. 創(chuàng)建kafka生產(chǎn)者配置對(duì)象
        Properties properties = new Properties();
        // 2. 給kafka的配置對(duì)象添加信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        // key, value初始化【必須有】
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        // =========> 添加自定義分區(qū)器 <============
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.example.kafkademo.producer.MyPartitioner")
        
        return properties;
    }
}