1
我试着写一些简单的生产者写入消息卡夫卡 我已经下载kafka0.7“ClassCastException异常:kafka.message.Message不能转换为java.lang.String”卡夫卡生产者
运行卡夫卡服务器,生产者和消费者
> bin/kafka-server-start.sh config/server.properties
> bin/kafka-console-consumer.sh --zookeeper zooserver:2181 --topic test --from-beginning
> bin/kafka-console-producer.sh --zookeeper zooserver:2181 --topic test
而且一切正常!
然后我创建了简单的消费者,它也有效。
SimpleConsumer consumer = new SimpleConsumer("172.21.110.134", 9092, 10000, 1024000);
…
ByteBufferMessageSet messages = consumer.fetch(fetchRequest);
System.out.println("consumed: " + Utils.toString(msg.message().payload(), "UTF-8"));
但接下来的生产者总是抛出错误
ProducerConfig config = new ProducerConfig(props);
…
Producer<String, Message> producer = new Producer<String, Message>(config);
producer.send(new ProducerData<String, Message>("test", new Message("Hello World".getBytes())));
收到以下异常
Exception in thread "main" java.lang.ClassCastException: kafka.message.Message cannot be cast to java.lang.String
at kafka.serializer.StringEncoder.toMessage(Encoder.scala:30)
at kafka.producer.ProducerPool$$anonfun$send$1$$anonfun$3$$anonfun$apply$1.apply(ProducerPool.scala:107)
我用Java代码的Maven依赖
<dependency>
<groupId>com.twitter</groupId>
<artifactId>kafka_2.9.2</artifactId>
<version>0.7.0</version>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.9.2</version>
</dependency>
P租赁帮助,我做错了什么?为什么我不能通过简单的java代码写入?
哇感谢,这么小的事! – Julias