kafka producerrecord如何發送

要使用Kafka ProducerRecord發送消息，您需要遵循以下步驟：

1. 添加Kafka客戶端依賴項：首先，確保在項目的pom.xml文件中添加了Kafka客戶端的依賴項。對于Maven項目，添加以下依賴項：

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

2. 創建Kafka生產者配置：創建一個Kafka生產者配置對象，指定Kafka集群的地址、密鑰生成器等配置信息。

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");

3. 創建Kafka生產者實例：使用配置對象創建一個Kafka生產者實例。

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

4. 創建Kafka ProducerRecord實例：創建一個Kafka ProducerRecord實例，指定主題、鍵和值。

String topic = "your_topic_name";
String key = "your_key";
String value = "your_value";
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

5. 發送Kafka ProducerRecord：使用Kafka生產者實例發送Kafka ProducerRecord。

producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println("Sent record to topic: " + metadata.topic() + ", partition: " + metadata.partition() + ", offset: " + metadata.offset());
        }
    }
});

6. 關閉Kafka生產者：在完成消息發送后，關閉Kafka生產者以釋放資源。

producer.close();

將以上代碼整合到一個完整的示例中：

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecordMetadata;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        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");

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

        String topic = "your_topic_name";
        String key = "your_key";
        String value = "your_value";
        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) {
                    exception.printStackTrace();
                } else {
                    System.out.println("Sent record to topic: " + metadata.topic() + ", partition: " + metadata.partition() + ", offset: " + metadata.offset());
                }
            }
        });

        producer.close();
    }
}

請注意，您需要根據實際情況替換your_topic_name、your_key和your_value。