在Java中調用Kafka接口發送數據,你可以使用Kafka的Java客戶端庫來實現。以下是一種常見的實現方式:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaProducerExample {
public static void main(String[] args) {
// 設置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生產者對象
Producer<String, String> producer = new KafkaProducer<>(props);
// 創建消息對象
String topic = "my-topic";
String key = "key1";
String value = "Hello, Kafka!";
// 發送消息
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("Error sending message: " + exception.getMessage());
} else {
System.out.println("Message sent successfully! Offset: " + metadata.offset());
}
}
});
// 關閉生產者
producer.close();
}
}
在上面的代碼中,你需要調整bootstrap.servers
屬性的值為Kafka集群的地址。你還可以調整其他配置屬性,如序列化器和分區策略。
這就是一個簡單的Java調用Kafka接口發送數據的例子。根據你的需求,你可以調整代碼來滿足更復雜的場景。