这里面我们介绍一下序列化框架arvo的使用，在kafka中使用的就是这个。
arvo的使用
一、需要通过插件生成Model类方式
一、生成我们的数据模型User.java
我们在resources里面定义即将要生成的User类的avsc结构。user.avsc的内容如下
{ "namespace": "com.linux.huhx.avro", "type": "record", "name": "User", "fields": [ { "name": "name", "type": "string" }, { "name": "favorite_number", "type": [ "int", "null" ] }, { "name": "favorite_color", "type": [ "string", "null" ] } ] }
然后我们在pom.xml添加依赖与avro-maven-plugin插件
<dependency> <groupId>org.apache.avro</groupId> <artifactId>avro</artifactId> <version>1.8.2</version> </dependency> <!-- add arvo plugin --> <plugin> <groupId>org.apache.avro</groupId> <artifactId>avro-maven-plugin</artifactId> <version>1.8.2</version> <executions> <execution> <phase>generate-sources</phase> <goals> <goal>schema</goal> </goals> <configuration> <sourceDirectory>${project.basedir}/src/main/resources</sourceDirectory> <outputDirectory>${project.basedir}/src/main/java</outputDirectory> </configuration> </execution> </executions> </plugin>
运行mvn generate-sources，就可以在arvo目录下面看到生成的User类。内容如下：
/** * Autogenerated by Avro * * DO NOT EDIT DIRECTLY */ package com.linux.huhx.avro; import org.apache.avro.specific.SpecificData; import org.apache.avro.message.BinaryMessageEncoder; import org.apache.avro.message.BinaryMessageDecoder; import org.apache.avro.message.SchemaStore; @SuppressWarnings("all") @org.apache.avro.specific.AvroGenerated public class User extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord { private static final long serialVersionUID = -1677175913369996238L; public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{"type":"record","name":"User","namespace":"com.linux.huhx.avro","fields":[{"name":"name","type":"string"},{"name":"favorite_number","type":["int","null"]},{"name":"favorite_color","type":["string","null"]}]}"); public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; } private static SpecificData MODEL$ = new SpecificData(); private static final BinaryMessageEncoder<User> ENCODER = new BinaryMessageEncoder<User>(MODEL$, SCHEMA$); private static final BinaryMessageDecoder<User> DECODER = new BinaryMessageDecoder<User>(MODEL$, SCHEMA$); /** * Return the BinaryMessageDecoder instance used by this class. */ public static BinaryMessageDecoder<User> getDecoder() { return DECODER; } /** * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}. * @param resolver a {@link SchemaStore} used to find schemas by fingerprint */ public static BinaryMessageDecoder<User> createDecoder(SchemaStore resolver) { return new BinaryMessageDecoder<User>(MODEL$, SCHEMA$, resolver); } /** Serializes this User to a ByteBuffer. */ public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException { return ENCODER.encode(this); } /** Deserializes a User from a ByteBuffer. */ public static User fromByteBuffer( java.nio.ByteBuffer b) throws java.io.IOException { return DECODER.decode(b); } @Deprecated public java.lang.CharSequence name; @Deprecated public java.lang.Integer favorite_number; @Deprecated public java.lang.CharSequence favorite_color; /** * Default constructor. Note that this does not initialize fields * to their default values from the schema. If that is desired then * one should use <code>newBuilder()</code>. */ public User() {} /** * All-args constructor. * @param name The new value for name * @param favorite_number The new value for favorite_number * @param favorite_color The new value for favorite_color */ public User(java.lang.CharSequence name, java.lang.Integer favorite_number, java.lang.CharSequence favorite_color) { this.name = name; this.favorite_number = favorite_number; this.favorite_color = favorite_color; } public org.apache.avro.Schema getSchema() { return SCHEMA$; } // Used by DatumWriter. Applications should not call. public java.lang.Object get(int field$) { switch (field$) { case 0: return name; case 1: return favorite_number; case 2: return favorite_color; default: throw new org.apache.avro.AvroRuntimeException("Bad index"); } } // Used by DatumReader. Applications should not call. @SuppressWarnings(value="unchecked") public void put(int field$, java.lang.Object value$) { switch (field$) { case 0: name = (java.lang.CharSequence)value$; break; case 1: favorite_number = (java.lang.Integer)value$; break; case 2: favorite_color = (java.lang.CharSequence)value$; break; default: throw new org.apache.avro.AvroRuntimeException("Bad index"); } } /** * Gets the value of the 'name' field. * @return The value of the 'name' field. */ public java.lang.CharSequence getName() { return name; } /** * Sets the value of the 'name' field. * @param value the value to set. */ public void setName(java.lang.CharSequence value) { this.name = value; } /** * Gets the value of the 'favorite_number' field. * @return The value of the 'favorite_number' field. */ public java.lang.Integer getFavoriteNumber() { return favorite_number; } /** * Sets the value of the 'favorite_number' field. * @param value the value to set. */ public void setFavoriteNumber(java.lang.Integer value) { this.favorite_number = value; } /** * Gets the value of the 'favorite_color' field. * @return The value of the 'favorite_color' field. */ public java.lang.CharSequence getFavoriteColor() { return favorite_color; } /** * Sets the value of the 'favorite_color' field. * @param value the value to set. */ public void setFavoriteColor(java.lang.CharSequence value) { this.favorite_color = value; } /** * Creates a new User RecordBuilder. * @return A new User RecordBuilder */ public static com.linux.huhx.avro.User.Builder newBuilder() { return new com.linux.huhx.avro.User.Builder(); } /** * Creates a new User RecordBuilder by copying an existing Builder. * @param other The existing builder to copy. * @return A new User RecordBuilder */ public static com.linux.huhx.avro.User.Builder newBuilder(com.linux.huhx.avro.User.Builder other) { return new com.linux.huhx.avro.User.Builder(other); } /** * Creates a new User RecordBuilder by copying an existing User instance. * @param other The existing instance to copy. * @return A new User RecordBuilder */ public static com.linux.huhx.avro.User.Builder newBuilder(com.linux.huhx.avro.User other) { return new com.linux.huhx.avro.User.Builder(other); } /** * RecordBuilder for User instances. */ public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<User> implements org.apache.avro.data.RecordBuilder<User> { private java.lang.CharSequence name; private java.lang.Integer favorite_number; private java.lang.CharSequence favorite_color; /** Creates a new Builder */ private Builder() { super(SCHEMA$); } /** * Creates a Builder by copying an existing Builder. * @param other The existing Builder to copy. */ private Builder(com.linux.huhx.avro.User.Builder other) { super(other); if (isValidValue(fields()[0], other.name)) { this.name = data().deepCopy(fields()[0].schema(), other.name); fieldSetFlags()[0] = true; } if (isValidValue(fields()[1], other.favorite_number)) { this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number); fieldSetFlags()[1] = true; } if (isValidValue(fields()[2], other.favorite_color)) { this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color); fieldSetFlags()[2] = true; } } /** * Creates a Builder by copying an existing User instance * @param other The existing instance to copy. */ private Builder(com.linux.huhx.avro.User other) { super(SCHEMA$); if (isValidValue(fields()[0], other.name)) { this.name = data().deepCopy(fields()[0].schema(), other.name); fieldSetFlags()[0] = true; } if (isValidValue(fields()[1], other.favorite_number)) { this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number); fieldSetFlags()[1] = true; } if (isValidValue(fields()[2], other.favorite_color)) { this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color); fieldSetFlags()[2] = true; } } /** * Gets the value of the 'name' field. * @return The value. */ public java.lang.CharSequence getName() { return name; } /** * Sets the value of the 'name' field. * @param value The value of 'name'. * @return This builder. */ public com.linux.huhx.avro.User.Builder setName(java.lang.CharSequence value) { validate(fields()[0], value); this.name = value; fieldSetFlags()[0] = true; return this; } /** * Checks whether the 'name' field has been set. * @return True if the 'name' field has been set, false otherwise. */ public boolean hasName() { return fieldSetFlags()[0]; } /** * Clears the value of the 'name' field. * @return This builder. */ public com.linux.huhx.avro.User.Builder clearName() { name = null; fieldSetFlags()[0] = false; return this; } /** * Gets the value of the 'favorite_number' field. * @return The value. */ public java.lang.Integer getFavoriteNumber() { return favorite_number; } /** * Sets the value of the 'favorite_number' field. * @param value The value of 'favorite_number'. * @return This builder. */ public com.linux.huhx.avro.User.Builder setFavoriteNumber(java.lang.Integer value) { validate(fields()[1], value); this.favorite_number = value; fieldSetFlags()[1] = true; return this; } /** * Checks whether the 'favorite_number' field has been set. * @return True if the 'favorite_number' field has been set, false otherwise. */ public boolean hasFavoriteNumber() { return fieldSetFlags()[1]; } /** * Clears the value of the 'favorite_number' field. * @return This builder. */ public com.linux.huhx.avro.User.Builder clearFavoriteNumber() { favorite_number = null; fieldSetFlags()[1] = false; return this; } /** * Gets the value of the 'favorite_color' field. * @return The value. */ public java.lang.CharSequence getFavoriteColor() { return favorite_color; } /** * Sets the value of the 'favorite_color' field. * @param value The value of 'favorite_color'. * @return This builder. */ public com.linux.huhx.avro.User.Builder setFavoriteColor(java.lang.CharSequence value) { validate(fields()[2], value); this.favorite_color = value; fieldSetFlags()[2] = true; return this; } /** * Checks whether the 'favorite_color' field has been set. * @return True if the 'favorite_color' field has been set, false otherwise. */ public boolean hasFavoriteColor() { return fieldSetFlags()[2]; } /** * Clears the value of the 'favorite_color' field. * @return This builder. */ public com.linux.huhx.avro.User.Builder clearFavoriteColor() { favorite_color = null; fieldSetFlags()[2] = false; return this; } @Override @SuppressWarnings("unchecked") public User build() { try { User record = new User(); record.name = fieldSetFlags()[0] ? this.name : (java.lang.CharSequence) defaultValue(fields()[0]); record.favorite_number = fieldSetFlags()[1] ? this.favorite_number : (java.lang.Integer) defaultValue(fields()[1]); record.favorite_color = fieldSetFlags()[2] ? this.favorite_color : (java.lang.CharSequence) defaultValue(fields()[2]); return record; } catch (java.lang.Exception e) { throw new org.apache.avro.AvroRuntimeException(e); } } } @SuppressWarnings("unchecked") private static final org.apache.avro.io.DatumWriter<User> WRITER$ = (org.apache.avro.io.DatumWriter<User>)MODEL$.createDatumWriter(SCHEMA$); @Override public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException { WRITER$.write(this, SpecificData.getEncoder(out)); } @SuppressWarnings("unchecked") private static final org.apache.avro.io.DatumReader<User> READER$ = (org.apache.avro.io.DatumReader<User>)MODEL$.createDatumReader(SCHEMA$); @Override public void readExternal(java.io.ObjectInput in) throws java.io.IOException { READER$.read(this, SpecificData.getDecoder(in)); } }
二、序列化我们的User类
public static void serializing() throws IOException { User user1 = new User(); user1.setName("Alyssa"); user1.setFavoriteNumber(256); User user2 = new User("Ben", 7, "red"); User user3 = User.newBuilder() .setName("Charlie") .setFavoriteColor("blue") .setFavoriteNumber(null) .build(); DatumWriter<User> userDatumWriter = new SpecificDatumWriter<>(User.class); DataFileWriter<User> dataFileWriter = new DataFileWriter<>(userDatumWriter); dataFileWriter.create(user1.getSchema(), new File("file/users.avro")); dataFileWriter.append(user1); dataFileWriter.append(user2); dataFileWriter.append(user3); dataFileWriter.close(); }
上面可以看到多种的创建User类的方式，运行上述代码会把User的有一个实例序列化到文件file/users.avro中。
三、反序列化我们的User类
public static void deserializing() throws IOException { // Deserialize Users from disk DatumReader<User> userDatumReader = new SpecificDatumReader<>(User.class); DataFileReader<User> dataFileReader = new DataFileReader<User>(new File("file/users.avro"), userDatumReader); User user = null; while (dataFileReader.hasNext()) { user = dataFileReader.next(user); System.out.println(user); } }
根据我们上述生成的序列化file/users.avro文件反序列化生成的User实例，可以看到如下的输出：
{"name": "Alyssa", "favorite_number": 256, "favorite_color": null}
{"name": "Ben", "favorite_number": 7, "favorite_color": "red"}
{"name": "Charlie", "favorite_number": null, "favorite_color": "blue"}
二、不需要生成User类的方式
Avro中的数据始终与其相应的模式一起存储，这意味着无论我们是否提前知道模式，我们总是可以读取序列化项目。这允许我们在不生成代码的情况下执行序列化和反序列化。如果是这种方式的话，pom.xml里面的arvo插件不是必须的。
一、序列化User类
public static void serializing() throws IOException { Schema schema = new Schema.Parser().parse(WithoutUserGen.class.getResourceAsStream("/user.avsc")); GenericRecord user1 = new GenericData.Record(schema); user1.put("name", "Alyssa"); user1.put("favorite_number", 256); // Leave favorite color null GenericRecord user2 = new GenericData.Record(schema); user2.put("name", "Ben"); user2.put("favorite_number", 7); user2.put("favorite_color", "red"); File file = new File("file/users2.avro"); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter); dataFileWriter.create(schema, file); dataFileWriter.append(user1); dataFileWriter.append(user2); dataFileWriter.close(); }
这里把两个User类的实例序列化到file/users2.avro文件中
二、反序列化User类
public static void deserializing() throws IOException { Schema schema = new Schema.Parser().parse(WithoutUserGen.class.getResourceAsStream("/user.avsc")); File file = new File("file/users2.avro"); DatumReader<GenericRecord> datumReader = new GenericDatumReader<>(schema); DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(file, datumReader); GenericRecord user = null; while (dataFileReader.hasNext()) { user = dataFileReader.next(user); System.out.println(user); } }
可以看到以下的输出
{"name": "Alyssa", "favorite_number": 256, "favorite_color": null}
{"name": "Ben", "favorite_number": 7, "favorite_color": "red"}