我正在使用 Avro 序列化对象,然后将它们添加到 Kafka 消息中,这些消息将由客户端使用和反序列化。我已经尝试了几种不同的序列化方法,但它们似乎都没有在数据流中嵌入模式。这是我的序列化代码的最新版本。您可以看到注释掉的尝试使用各种可用的编写器。
public static byte[] Serialize<T>(T recordObj) where T : ISpecificRecord
{
Log.Info("Serializing {0} object to Avro.", typeof(T));
try
{
using (var ms = new MemoryStream())
{
var encoder = new BinaryEncoder(ms);
//var writer = new SpecificDefaultWriter(recordObj.Schema);
var writer = new SpecificDatumWriter<T>(recordObj.Schema);
//writer.Write(recordObj.Schema, recordObj, encoder);
writer.Write(recordObj, encoder);
return ms.ToArray();
}
}
catch (Exception ex)
{
Log.Error("Failed to Avro serialize object. {0}", ex);
return null;
}
}
Run Code Online (Sandbox Code Playgroud)
我不确定还有什么要尝试的。
在深入研究了实际的 Avro 代码后,我发现我需要一个 FileWriter,但无法弄清楚如何实例化一个,因为DataFileWriter没有公共构造函数。原来在 DataFileWriter 类上有一个名为 OpenWriter 的静态方法,它接受一个 DatumWriter 和一个 Stream 并返回一个 DataFileWriter。下面的代码现在正确地在结果数据流中包含了对象元数据。
public static byte[] Serialize<T>(T recordObj) where T : ISpecificRecord
{
Log.Info("Serializing {0} object to Avro.",typeof(T));
try
{
using(var ms = new MemoryStream())
{
var specDatumWriter = new SpecificDatumWriter<T>(recordObj.Schema);
var specDataWriter = Avro.File.DataFileWriter<T>.OpenWriter(specDatumWriter, ms);
specDataWriter.Append(recordObj);
specDataWriter.Flush();
specDataWriter.Close();
return ms.ToArray();
}
}
catch(Exception ex)
{
Log.Error("Failed to Avro serialize object. {0}",ex);
return null;
}
}
Run Code Online (Sandbox Code Playgroud)