Kafka生产者如何管理连接

Kafka生产者可以使用连接池来管理与Kafka集群的连接。连接池可以重用已建立的连接，避免每次发送消息时都需要建立新的连接。下面是一个使用连接池的Kafka生产者示例代码：

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

import java.util.Properties;

public class KafkaProducerExample {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC = "my-topic";

    public static void main(String[] args) {
        // 创建Kafka生产者配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 创建连接池
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 发送消息
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "key-" + i, "value-" + i);
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            System.err.println("Error sending message: " + exception.getMessage());
                        } else {
                            System.out.println("Message sent successfully! Topic: " + metadata.topic() +
                                    ", Partition: " + metadata.partition() + ", Offset: " + metadata.offset());
                        }
                    }
                });
            }
        } finally {
            // 关闭连接池
            producer.close();
        }
    }
}

在上面的示例中，创建了一个KafkaProducer实例，并使用连接池来管理与Kafka集群的连接。KafkaProducer类提供了发送消息的方法send，可以在发送消息时指定回调函数Callback，用于处理消息发送结果。在最后，通过调用close方法关闭连接池。