As everyone knows the kafka producer can send message to specific patition:

producer.send(new ProducerRecord<String, String>("topic", patition,"key","value"));

Now learn to receive message from kafka consumer
just simple configuration:

1、config:
配置指定消费0和1两个partition:
partition=0,1

2、

//get partition conf
String partition = props.getProperty("partition");
//have specific patition:
if (partition != null) {
    List<TopicPartition> topicPartitions = new ArrayList<>();
    String[] partitions = partition.split(",");
    for (int i = 0; i < partitions.length; i++) {
        TopicPartition topicPartition = new TopicPartition(topic, Integer.parseInt(partitions[i]));
        topicPartitions.add(topicPartition);
    }
    consumer.assign(topicPartitions);
//do not specific patition :
} else {
    this.consumer.subscribe(Arrays.asList(topic));
}

指定partition时使用consumer.assign() 即用户自定义partition分配策略。

未指定时使用consumer.subscribe()即消费模式为指定topic,由kafka指定partition分配策略。

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐