scala + kafka 不发送带有 id 的消息答案

【问题标题】：scala + kafka does not send message with idscala + kafka 不发送带有 id 的消息
【发布时间】：2015-08-11 16:07:27
【问题描述】：

我正在尝试通过我的演员向 Kafka 发送消息，但它不起作用。

以下代码有效

new KeyedMessage[String, Array[Byte]]("my-topic", msg.message)

这个不...为什么？

new KeyedMessage[String, Array[Byte]]("my-topic", msg.id, msg.message)

甚至

new KeyedMessage[String, Array[Byte]]("my-topic", msg.id, null, msg.message)

将partition设置为null，强制只填充id，但还是一样。

有什么想法吗？

编辑

好吧，只是意识到当以字节数组发送消息时它不起作用。

我已经更改了我的代码以允许以字符串形式发送消息并且它可以工作。

以下内容不起作用：

  props.put("serializer.class", "kafka.serializer.DefaultEncoder")

  override def receive: Receive = {
    case msg: Message =>
      val keyedMessage = new KeyedMessage[String, Array[Byte]]("my-topic", msg.id, msg.message)
      producer.send(keyedMessage)

    case _ => log.error("Got a msg that I don't understand")
  }

将 Array[Byte] 修改为 String，同时将 DefaultEncoder 修改为 StringEncoder，就可以了。

有什么想法吗？

【问题讨论】：

标签： scala akka apache-kafka

【解决方案1】：

嗯，我刚刚发现了问题。

实际上，编码器有两种选择。 DefaultEncoder 和 StringEncoder。我试图使用默认编码器将我的 id 作为字符串发送，并将消息作为字节数组发送。

查看Encoder.scala，可以看到DefaultEncoder的实现和答案。

/**
 * The default implementation is a no-op, it just returns the same array it takes in
 */
class DefaultEncoder(props: VerifiableProperties = null) extends Encoder[Array[Byte]] {
  override def toBytes(value: Array[Byte]): Array[Byte] = value
}

它只是返回字节数组，因为我传递的是一个字符串，所以它抛出了强制转换异常。

【讨论】：