阅读量:0
使用Kafka的ProducerAPI可以向一个或多个Topic发送消息。
首先,需要创建一个Producer实例,并将Kafka集群的地址传递给它。可以通过指定一个或多个bootstrap.servers属性来指定Kafka集群的地址。
Properties props = new Properties(); props.put("bootstrap.servers", "kafka1:9092,kafka2:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer<String, String> producer = new KafkaProducer<>(props);
接下来,可以使用Producer的send方法将消息发送到Topic。需要创建一个ProducerRecord实例,指定要发送的消息的Topic、消息的键和值。
String topic = "my-topic"; String key = "my-key"; String value = "Hello, Kafka!"; ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value); producer.send(record);
最后,使用完Producer后需要调用close方法来关闭Producer并释放相关资源。
producer.close();
以上代码片段展示了如何向一个Topic发送一条消息。可以根据需求进行调整,例如发送多条消息、指定消息的分区、添加消息的回调函数等。