public interface Partitioner extends Configurable, Closeable {
    int partition(String topic, Object key, byte[] keyBytes,
                  Object value, byte[] valueBytes, Cluster cluster);
    void close();
    void configure(Map<String, ?> configs);
}

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class UserIdPartitioner implements Partitioner {
    // VIP 用戶列表（可從配置或數(shù)據(jù)庫(kù)加載）
    private static final Set<String> VIP_USERS = Set.of("1001", "2005", "9999");
    // 緩存 VIP 用戶的專屬分區(qū)，避免頻繁計(jì)算
    private final Map<String, Integer> vipPartitionCache = new ConcurrentHashMap<>();
    @Override
    public void configure(Map<String, ?> configs) {
        // 可在此處讀取自定義配置，如 VIP 列表、分區(qū)偏移量等
        System.out.println("UserIdPartitioner configured.");
    }
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 獲取該 topic 的所有分區(qū)信息
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (key == null) {
            // 無 key 時(shí)，使用輪詢（簡(jiǎn)化版）
            return Math.abs(key.hashCode()) % numPartitions;
        }
        String userId = key.toString();
        if (VIP_USERS.contains(userId)) {
            // VIP 用戶：固定分配到前 N 個(gè)分區(qū)（例如前 2 個(gè)）
            // 為每個(gè) VIP 用戶分配唯一分區(qū)，避免沖突
            return vipPartitionCache.computeIfAbsent(userId, k -> {
                int index = VIP_USERS.stream().toList().indexOf(k);
                return index % Math.min(2, numPartitions); // 最多使用 2 個(gè)分區(qū)
            });
        } else {
            // 普通用戶：使用標(biāo)準(zhǔn)哈希
            return Math.abs(userId.hashCode()) % numPartitions;
        }
    }
    @Override
    public void close() {
        vipPartitionCache.clear();
        System.out.println("UserIdPartitioner closed.");
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 關(guān)鍵配置：指定自定義分區(qū)器
props.put("partitioner.class", "com.example.UserIdPartitioner");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// 發(fā)送消息
producer.send(new ProducerRecord<>("user-events", "1001", "VIP login"));
producer.send(new ProducerRecord<>("user-events", "5001", "Normal user action"));

public class LoadAwarePartitioner implements Partitioner {
    private final Map<Integer, Long> partitionLoad = new ConcurrentHashMap<>();
    private final Random random = new Random();
    @Override
    public void configure(Map<String, ?> configs) {}
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (key == null) {
            // 無 key：選擇負(fù)載最低的分區(qū)
            return findLeastLoadedPartition(numPartitions);
        }
        // 有 key：仍需保證相同 key 進(jìn)同一分區(qū)（順序性）
        // 但我們可以記錄該分區(qū)的負(fù)載
        int targetPartition = Math.abs(key.hashCode()) % numPartitions;
        updateLoad(targetPartition);
        return targetPartition;
    }
    private int findLeastLoadedPartition(int numPartitions) {
        long minLoad = Long.MAX_VALUE;
        int bestPartition = 0;
        for (int i = 0; i < numPartitions; i++) {
            long load = partitionLoad.getOrDefault(i, 0L);
            if (load < minLoad) {
                minLoad = load;
                bestPartition = i;
            }
        }
        // 更新負(fù)載（模擬）
        updateLoad(bestPartition);
        return bestPartition;
    }
    private void updateLoad(int partition) {
        partitionLoad.compute(partition, (k, v) -> (v == null) ? 1L : v + 1);
    }
    @Override
    public void close() {
        partitionLoad.clear();
    }
}

// ? 錯(cuò)誤！相同 key 可能進(jìn)入不同分區(qū)
public int partition(...) {
    if (isVip(key)) {
        return random.nextInt(numPartitions); // 隨機(jī)分配 VIP
    }
    return hash(key) % numPartitions;
}

// ? 正確：相同 key 始終進(jìn)入同一分區(qū)
public int partition(...) {
    if (isVip(key)) {
        // 所有 VIP 用戶進(jìn)入分區(qū) 0
        return 0;
    }
    return hash(key) % numPartitions;
}

// ? 使用 ConcurrentHashMap
private final Map<String, Integer> cache = new ConcurrentHashMap<>();
// ? 非線程安全
private Map<String, Integer> cache = new HashMap<>();

@Test
public void testVipUserRouting() {
    UserIdPartitioner partitioner = new UserIdPartitioner();
    partitioner.configure(Collections.emptyMap());
    Cluster cluster = mockClusterWithPartitions(4); // 模擬 4 分區(qū)集群
    int p1 = partitioner.partition("test", "1001", null, null, null, cluster);
    int p2 = partitioner.partition("test", "1001", null, null, null, cluster);
    assertEquals(p1, p2); // 同一 VIP 應(yīng)進(jìn)入同一分區(qū)
    assertTrue(p1 < 2);   // 且應(yīng)在前 2 個(gè)分區(qū)
}

{
  "order_id": "O12345",
  "user_id": "U789",
  "region": "CN-EAST",
  "amount": 299.99
}

public class RegionAwareOrderPartitioner implements Partitioner {
    private static final Map<String, Integer> REGION_PARTITION_OFFSET = Map.of(
        "CN-EAST", 0,
        "CN-NORTH", 2,
        "CN-SOUTH", 4
    );
    private static final int PARTITIONS_PER_REGION = 2;
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 假設(shè) value 是 JSON 字符串
        String jsonValue = (String) value;
        String region = extractRegionFromJson(jsonValue); // 解析 region
        int basePartition = REGION_PARTITION_OFFSET.getOrDefault(region, 0);
        int totalPartitions = cluster.partitionsForTopic(topic).size();
        // 計(jì)算該 region 的可用分區(qū)范圍
        int start = basePartition;
        int end = Math.min(basePartition + PARTITIONS_PER_REGION, totalPartitions);
        if (start >= totalPartitions) {
            // fallback to default
            return Math.abs(key.hashCode()) % totalPartitions;
        }
        // 在 region 內(nèi)部按 user_id 哈希
        int userHash = Math.abs(key.hashCode());
        int regionPartition = start + (userHash % (end - start));
        return regionPartition;
    }
    private String extractRegionFromJson(String json) {
        // 簡(jiǎn)化：實(shí)際應(yīng)使用 JSON 解析庫(kù)
        int start = json.indexOf("\"region\":\"") + 11;
        int end = json.indexOf("\"", start);
        return json.substring(start, end);
    }
    @Override
    public void configure(Map<String, ?> configs) {}
    @Override
    public void close() {}
}

渲染錯(cuò)誤: Mermaid 渲染失敗: Parsing failed: unexpected character: ->“<- at offset: 29, skipped 6 characters. unexpected character: ->：<- at offset: 36, skipped 1 characters. unexpected character: ->“<- at offset: 44, skipped 6 characters. unexpected character: ->：<- at offset: 51, skipped 1 characters. unexpected character: ->“<- at offset: 59, skipped 6 characters. unexpected character: ->：<- at offset: 66, skipped 1 characters. Expecting token of type 'EOF' but found `2`. Expecting token of type 'EOF' but found `2`. Expecting token of type 'EOF' but found `2`.

kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic user-events

user-events:0:15000
user-events:1:5000
user-events:2:5000