對于這個小案例，簡單來個圖示，

友情補充pom

<dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.3.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.11</artifactId><version>2.3.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.3.1</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.47</version></dependency><!-- 向kafka 生產(chǎn)數(shù)據(jù)需要包 --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>0.10.0.0</version></dependency></dependencies>

代碼

package com.kafkaimport java.util.Propertiesimport org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}object KafkaDemo {def main(args: Array[String]): Unit = {val props = new Properties()props.setProperty("bootstrap.servers","henu2:9092")props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");val kp = new KafkaProducer[String,String](props)println("++++")kp.send(new ProducerRecord[String,String]("george","georgedage"))kp.close()} }

運行前，在linux上進行kafka的消費者命令。

kafka-console-consumer.sh --bootstrap-server henu2:9092 --from-beginning --topic george

然后運行idea的代碼，注意你的主題名稱要一致。

運行結(jié)果：

總結(jié)

以上是生活随笔為你收集整理的使用idea编写代码作为生产者，Kafka接收其发来的信息【小案例】(一)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。