生產和消費卡夫卡Avro郵件沒有匯合組件

Question

我想找到一個例子，我可以製作和訂閱卡夫卡的avro郵件。

在這個時候，我想使用沒有任何融合附加組件的「香草」卡夫卡部署。

這是可能的嗎？我發現的所有例子都很快就開始使用融合特定工具來處理avro消息。

我相信我應該有一種方法可以在kafka平臺上發佈和使用avro消息，並使用任何「分發特定」無插件。

Answer 1

當然，你可以做到這一點，沒有任何Confluent工具。但是你必須在你身邊做額外的工作（例如在你的應用程序代碼中） - 這是提供Avro相關工具的最初動機，例如你提到的來自Confluent的工具。

一種方法是直接使用Apache Avro Java API手動序列化/反序列化Kafka消息的有效負載（例如從YourJavaPojo到byte[]）。 （我想你認爲Java是選擇的編程語言。）這將如何？這是一個例子。

• 首先，您將手動序列化應用程序中的數據有效載荷，將數據寫入卡夫卡。在這裏，您可以使用Avro序列化API對有效負載進行編碼（從Java POjo到byte[]），然後使用Kafka的Java生產者客戶端將寫入編碼負載寫入Kafka主題。
• 然後，在數據管道的下游，您將在另一個從Kafka讀取數據的應用程序中進行反序列化。在這裏，您可以使用Kafka的Java客戶端客戶端讀取來自同一個Kafka主題的（編碼）數據，並使用Avro反序列化API再次解碼有效載荷（從byte[]到Java pojo）。

當然，在使用流處理工具如Kafka Streams（將包含在即將推出的Apache Kafka 0.10中）或Apache Storm中時，您也可以直接使用Avro API。

最後，您還可以選擇使用某些實用程序庫（無論是來自Confluent還是其他地方），以便您不必直接使用Apache Avro API。關於它的價值，我已經在kafka-storm-starter，例如，如AvroDecoderBolt.scala所示。在這裏，Avro序列化/反序列化通過使用Scala庫Twitter Bijection完成。下面是AvroDecoderBolt.scala一個例子片斷給你的總體思路：

// This tells Bijection how to automagically deserialize a Java type `T`, 
    // given a byte array `byte[]`. 
    implicit private val specificAvroBinaryInjection: Injection[T, Array[Byte]] = 
SpecificAvroCodecs.toBinary[T] 

    // Let's put Bijection to use. 
    private def decodeAndEmit(bytes: Array[Byte], collector: BasicOutputCollector) { 
    require(bytes != null, "bytes must not be null") 
    val decodeTry = Injection.invert(bytes) // <-- deserialization, using Twitter Bijection, happens here 
    decodeTry match { 
     case Success(pojo) => 
     log.debug("Binary data decoded into pojo: " + pojo) 
     collector.emit(new Values(pojo)) // <-- Here we are telling Storm to send the decoded payload to downstream consumers 
     () 
     case Failure(e) => log.error("Could not decode binary data: " + Throwables.getStackTraceAsString(e)) 
    } 
    } 

所以，是的，你當然可以選擇不使用任何額外的庫如匯合的Avro的串行器/解串器（目前提供作爲confluentinc/schema-registry部分）或Twitter's Bijection。是否值得額外的努力取決於你自己決定。