要使用Kafka ProducerRecord发送消息,您需要遵循以下步骤:
- 添加Kafka客户端依赖项:首先,确保在项目的pom.xml文件中添加了Kafka客户端的依赖项。对于Maven项目,添加以下依赖项:
org.apache.kafka kafka-clients 2.8.0
- 创建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");
- 创建Kafka生产者实例:使用配置对象创建一个Kafka生产者实例。
Producerproducer = new KafkaProducer<>(props);
- 创建Kafka ProducerRecord实例:创建一个Kafka ProducerRecord实例,指定主题、键和值。
String topic = "your_topic_name"; String key = "your_key"; String value = "https://www.yisu.com/ask/your_value"; ProducerRecordrecord = new ProducerRecord<>(topic, key, value);
- 发送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()); } } });
- 关闭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"); Producerproducer = new KafkaProducer<>(props); String topic = "your_topic_name"; String key = "your_key"; String value = "https://www.yisu.com/ask/your_value"; ProducerRecord 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
。