Kafka如何处理与分区相关的有键消息

在Kafka中，与分区相关的有键消息是通过键来决定消息将被分配到哪个分区中。下面是一个处理与分区相关的有键消息的Kafka代码示例：

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

public class KafkaPartitionedKeyMessageExample {

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String topic = "partitioned-topic";
        String key = "my-key";
        String value = "Hello Kafka!";

        // 生产者代码示例
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<>(producerProps);

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("Failed to send message: " + exception.getMessage());
                } else {
                    System.out.println("Message sent successfully to partition " + metadata.partition());
                }
            }
        });

        producer.close();

        // 消费者代码示例
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");

        Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> consumerRecord : records) {
                System.out.println("Received message: key = " + consumerRecord.key() + ", value = " + consumerRecord.value() + ", partition = " + consumerRecord.partition());
            }
        }
    }
}

在上述代码示例中，生产者使用ProducerRecord类来创建一个消息对象，并指定消息的键（key）、值（value）和目标主题（topic）。生产者将消息发送到Kafka集群，并通过回调函数处理消息发送结果。

消费者使用KafkaConsumer类来订阅主题，并使用poll方法来拉取消息。通过遍历ConsumerRecords对象来处理接收到的消息。在处理每个ConsumerRecord时，可以通过partition方法获取消息所属的分区。

这样，你就可以使用以上代码示例来处理与分区相关的有键消息。