Java 中间件Kafka 分区策略(自定义分区器实现负载均衡)

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 用户列表（可从配置或数据库加载）
    private static final set<string> vip_users = set.of("1001", "2005", "9999");
    // 缓存 vip 用户的专属分区，避免频繁计算
    private final map<string, integer> vippartitioncache = new concurrenthashmap<>();
    @override
    public void configure(map<string, ?> configs) {
        // 可在此处读取自定义配置，如 vip 列表、分区偏移量等
        system.out.println("useridpartitioner configured.");
    }
    @override
    public int partition(string topic, object key, byte[] keybytes,
                         object value, byte[] valuebytes, cluster cluster) {
        // 获取该 topic 的所有分区信息
        list<partitioninfo> partitions = cluster.partitionsfortopic(topic);
        int numpartitions = partitions.size();
        if (key == null) {
            // 无 key 时，使用轮询（简化版）
            return math.abs(key.hashcode()) % numpartitions;
        }
        string userid = key.tostring();
        if (vip_users.contains(userid)) {
            // vip 用户：固定分配到前 n 个分区（例如前 2 个）
            // 为每个 vip 用户分配唯一分区，避免冲突
            return vippartitioncache.computeifabsent(userid, k -> {
                int index = vip_users.stream().tolist().indexof(k);
                return index % math.min(2, numpartitions); // 最多使用 2 个分区
            });
        } else {
            // 普通用户：使用标准哈希
            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");
// 关键配置：指定自定义分区器
props.put("partitioner.class", "com.example.useridpartitioner");
kafkaproducer<string, string> producer = new kafkaproducer<>(props);
// 发送消息
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：选择负载最低的分区
            return findleastloadedpartition(numpartitions);
        }
        // 有 key：仍需保证相同 key 进同一分区（顺序性）
        // 但我们可以记录该分区的负载
        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;
            }
        }
        // 更新负载（模拟）
        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();
    }
}

// ❌ 错误！相同 key 可能进入不同分区
public int partition(...) {
    if (isvip(key)) {
        return random.nextint(numpartitions); // 随机分配 vip
    }
    return hash(key) % numpartitions;
}

// ✅ 正确：相同 key 始终进入同一分区
public int partition(...) {
    if (isvip(key)) {
        // 所有 vip 用户进入分区 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 分区集群
    int p1 = partitioner.partition("test", "1001", null, null, null, cluster);
    int p2 = partitioner.partition("test", "1001", null, null, null, cluster);
    assertequals(p1, p2); // 同一 vip 应进入同一分区
    asserttrue(p1 < 2);   // 且应在前 2 个分区
}

{
  "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) {
        // 假设 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();
        // 计算该 region 的可用分区范围
        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 内部按 user_id 哈希
        int userhash = math.abs(key.hashcode());
        int regionpartition = start + (userhash % (end - start));
        return regionpartition;
    }
    private string extractregionfromjson(string json) {
        // 简化：实际应使用 json 解析库
        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() {}
}

渲染错误: 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