我正在尝试配置我的kafkaconsumer,以便通过“kafkatopic”从kafkaproducer读取数据。
我的scala代码是:
package com.sundogsoftware.sparkstreaming
import java.util
import java.util.Properties
import org.apache.kafka.clients.consumer.KafkaConsumer
import scala.collection.JavaConverters._
import org.apache.kafka.clients.consumer.{ ConsumerRecords, KafkaConsumer }
import org.apache.kafka.clients.producer.{ KafkaProducer, Producer, ProducerConfig, ProducerRecord }
import java.util.Arrays;
object ConsumerExample extends App {
val props = new Properties()
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("partition.assignment.strategy", "range");
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(Arrays.asList("kafkatopic"))
while (true) {
val records = consumer.poll(100)
println(consumer)
println(records)
for (record <- records.asScala) {
println(record)
}
}
}
错误出现在consumer.subscribe(arrays.aslist(“kafkatopic”)):
重载方法值subscribe with alternatives:(x$1:org.apache.kafka.common.topicpartition*)unit(x$1:string*)unit不能应用于(java.util.list[string])
1条答案
按热度按时间wwodge7n1#
在scala中,通常按以下方式进行: