パブリッシャーを使用してメッセージを生成するには（reactive-kafkaで）？

Question

私は反応性カフカモジュールからのコードで動作する必要があります。https://github.com/akka/reactive-kafka/blob/master/README.md

私のコードで始まる：

val kafka = new ReactiveKafka() 

val kafkaIdpsMsgs: Publisher[StringKafkaMessage] = kafka.consume(
     ConsumerProperties(
     brokerList = kafkaHosts, 
     zooKeeperHost = zkHosts, 
     topic = "test", 
     groupId = "idps-translator", 
     decoder = new StringDecoder() 
    ).readFromEndOfStream()) 

    val kafkaSamples: Subscriber[String] = kafka.publish(ProducerProperties(
     brokerList = kafkaHosts, 
     topic = "test", 
     encoder = new StringEncoder() 
    )) 

私は（出版社による）メッセージを生成します。それを実現させるために書かなければならないコードは何ですか？

Answer 1

val done = Source(1 to 100) 
    .map(_.toString) 
    .map { elem => 
    new ProducerRecord[Array[Byte], String]("topic1", elem) 
    } 
    .runWith(Producer.plainSink(producerSettings)) 

あなたはこの文書を通過することができます：http://doc.akka.io/docs/akka-stream-kafka/current/producer.html