[Scala/Flink] 读写 Kafka

Flink 与 Kafka 的一些方法调用跟其版本有关, 下面是我使用的版本

  8
  8
  8

  1.12.1
  2.12.7
  2.4.1



  
     org.scala-lang
     scala-library
     ${scala.version}
   
   
   
      org.apache.flink
      flink-connector-kafka_2.12
      ${flink.version}
    
    
    
      org.apache.flink
      flink-scala_2.12
      ${flink.version}
	
	
	
      org.apache.flink
      flink-streaming-scala_2.12
      ${flink.version}
    

	
      org.apache.flink
      flink-clients_2.12
      ${flink.version}
    

	
        org.apache.kafka
        kafka-clients
        ${kafka.version}
    

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.datastream.DataStreamSource
import java.util.Properties

Object Test {
	def main(args: Array[String]): Unit = {
		val env = StreamExecutionEnvironment.getExecutionEnvironment
		val kafkaSubTopic = "topic-test"
		val kafkaGroupId = "group-test"
		val stream: DataStreamSource[String] = readKafka(env, kafkaSubTopic, kafkaGroupId, "latest")
		stream.print()
		env.execute()
	}
	
	def readKafka(
               env: StreamExecutionEnvironment,
               topic: String,
               groupId: String,
               offset: String = "latest"
            )= {
		props.put("bootstrap.servers", "{ip}:9092") // {ip} 替换为本机 ip
	    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
	    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
	    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringSerializer")
	    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
		props.put("auto.offset.reset", offset)
	
	    env.addSource(new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), props))
    }
}

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, Callback, RecordMetadata}
import com.alibaba.fastjson.JSON
import java.util.Properties

Object Test {
	def main(args: Array[String]): Unit = {
		val producer = new MyKafkaProducer("Test")
		val data = new JSONObject()
		data.put("a", "1")
		producer.send(data.toString())
	}

	class MyKafkaProducer(val topic: String) {
	    private val props = getProps()
	    private val producer = new KafkaProducer[String, String](props)
	
	    def send(data: String): Unit = {
	      val record = new ProducerRecord[String, String](topic, data)
	      
	      producer.send(record, new ProducerCallBack)
	    }
	
	    class ProducerCallBack extends Callback {
	      def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
	        if (exception != null) exception.printStackTrace()
	      }
	    }
    }
}