Akka Kafka 自定义序列化器答案

【问题标题】：Akka Kafka Custom SerializerAkka Kafka 自定义序列化器
【发布时间】：2018-01-07 07:28:30
【问题描述】：

我正在使用 Akka Kafka (Scala) 并希望发送自定义对象。

class TweetsSerializer extends Serializer[Seq[MyCustomType]] {

override def configure(configs: util.Map[String, _], isKey: Boolean):   Unit = ???

override def serialize(topic: String, data: Seq[MyCustomType]): Array[Byte] = ???

override def close(): Unit = ???

}

如何正确编写自己的序列化程序？还有，我应该如何处理字段config？

【问题讨论】：

只是谷歌示例？
@BranislavLazic 我在谷歌中找不到任何有用的例子:(

标签： scala serialization apache-kafka akka

【解决方案1】：

我会使用 StringSerializer，我的意思是，我会在生成它们之前将所有类型转换为字符串。然而，这行得通：

case class MyCustomType(a: Int)

  class TweetsSerializer extends Serializer[Seq[MyCustomType]] {

    private var encoding = "UTF8"

    override def configure(configs: java.util.Map[String, _], isKey: Boolean):   Unit = {
      val propertyName = if (isKey) "key.serializer.encoding"
      else "value.serializer.encoding"
      var encodingValue = configs.get(propertyName)
      if (encodingValue == null) encodingValue = configs.get("serializer.encoding")
      if (encodingValue != null && encodingValue.isInstanceOf[String]) encoding = encodingValue.asInstanceOf[String]
    }

    override def serialize(topic: String, data: Seq[MyCustomType]): Array[Byte] =
      try
          if (data == null) return null
          else return {
            data.map { v =>
              v.a.toString
            }
            .mkString("").getBytes("UTF8")
          }
      catch {
        case e: UnsupportedEncodingException =>
          throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding)
      }

    override def close(): Unit = Unit

  }

}

object testCustomKafkaSerializer extends App {


  implicit val producerConfig = {
    val props = new Properties()
    props.setProperty("bootstrap.servers", "localhost:9092")
    props.setProperty("key.serializer", classOf[StringSerializer].getName)
    props.setProperty("value.serializer", classOf[TweetsSerializer].getName)
    props
  }

  lazy val kafkaProducer = new KafkaProducer[String, Seq[MyCustomType]](producerConfig)

  // Create scala future from Java
  private def publishToKafka(id: String, data: Seq[MyCustomType]) = {
      kafkaProducer
        .send(new ProducerRecord("outTopic", id, data))
        .get()
  }

  val input = MyCustomType(1)

  publishToKafka("customSerializerTopic", Seq(input))


}

【讨论】：