/**

  * Created by soyo on 17-10-17.

  * 运行kafka程序 1.需要启动Zookeeper服务:./bin/zookeeper-server-start.sh config/zookeeper.properties

  *  2.启动Kafka服务：./bin/kafka-server-start.sh config/server.properties

  *  3.执行 DStream_Kafa_Producer

  *  4.执行 DStream_Kafa_Consumer

  */

object DStream_Kafa_Producer {

  def main(args: Array[String]): Unit = {

    val brokers="localhost:9092"

    val topic="wordsender"

    val messagePerSec=    //行数

    val wordsPerMessage=  //列数

    //配置Zookeeper

    val props= new util.HashMap[String,Object]()

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokers)

    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")

    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")

    val producer=new KafkaProducer[String,String](props)

    while (true){

      ( to messagePerSec).foreach({x=>

        val str=( to wordsPerMessage).map(x=>scala.util.Random.nextInt().toString).mkString(" ")

        println(str)

        val message=new ProducerRecord[String,String](topic,null,str)

        producer.send(message)

        val s2="ni hao 我的 测试 的 字符串"   //可以与message一起发送

        val message2=new ProducerRecord[String,String](topic,null,s2)

         producer.send(message2)

      })

      Thread.sleep()

    }

  }

}