|
|
kafkawriter_Scala样例代码是一段使用Apache Kafka的Scala编程语言编写的示例代码。该代码演示了如何使用Kafka生产者将消息发送到Kafka集群中的特定主题。它包括创建生产者实例、定义要发送的消息以及调用send方法将消息发送到Kafka集群的过程。
KafkaWriter 是一个用于将数据写入 Kafka 的 Scala 类，以下是一个简单的 KafkaWriter 样例代码，包括创建 KafkaWriter 实例、配置参数以及发送消息的方法。
zbhjijwkosdqinp.png
（图片来源网络，侵删）
import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
class KafkaWriter(brokers: String, topic: String) {
private val props = new Properties()
props.put("bootstrap.servers", brokers)
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
private val producer = new KafkaProducer[String, String](props)
def sendMessage(key: String, value: String): Unit = {
val record = new ProducerRecord[String, String](topic, key, value)
producer.send(record)
}
def close(): Unit = {
producer.close()
}
}
object KafkaWriterExample {
def main(args: Array[String]): Unit = {
val kafkaWriter = new KafkaWriter("localhost:9092", "testtopic")
// 发送消息
kafkaWriter.sendMessage("key1", "value1")
kafkaWriter.sendMessage("key2", "value2")
// 关闭生产者
kafkaWriter.close()
}
}
在这个示例中，我们首先导入了所需的 Kafka 相关类，我们创建了一个名为KafkaWriter 的类，它接受两个参数：brokers（Kafka 集群地址）和topic（要写入的主题）。
在KafkaWriter 类中，我们定义了一个名为props 的属性对象，用于存储 Kafka 生产者的配置信息，我们创建了一个名为producer 的 KafkaProducer 实例，并使用props 作为配置参数。
我们还定义了一个名为sendMessage 的方法，该方法接受一个键和一个值作为参数，并将它们封装在一个ProducerRecord 对象中，我们使用producer.send() 方法将记录发送到 Kafka。
我们定义了一个名为close 的方法，用于关闭 KafkaProducer 实例。
在KafkaWriterExample 对象中，我们创建了一个KafkaWriter 实例，并向其发送了两条消息，我们调用close 方法关闭生产者。
zbhj4byq3qquwyy.jpg
（图片来源网络，侵删） |
|
