一个简单的KafkaProducer(scala版)

object AmieKafkaProducer extends App{
  val producerProperties = new Properties()
  producerProperties.put("serializer.class", "kafka.serializer.StringEncoder")
  producerProperties.put("metadata.broker.list", "hdp01:9092")
  producerProperties.put("request.required.acks", "1")
  producerProperties.put("partitioner.class", "org.amie.spark.kafka.MyLogPartitioner")
  private val config: ProducerConfig = new ProducerConfig(producerProperties)
  private val producer: Producer[String, String] = new Producer(config)
  var id : Int =0
  while(true){
    val message = new KeyedMessage("amieTopic1",id.toString,"hello amie hello scala hello spark")
    ///kafka 中每一个message 都是keyedMessage, 即key value 类型. ---> key 可以被kafka 用来做负载均衡
    producer.send(message)
    id +=1
  }
}