如何使用HTTP作为传输读取通过Avro序列化的二进制数据

Question

我正在使用Avro作为序列化协议。我的服务已准备就绪，每个序列化/反序列化在内存中都可以正常工作。所以现在我想测试它，看看它在 HTTP 传输后是否正常工作。我认为编写一个测试方法很简单，但过了一段时间，我无法弄清楚，这是我尝试过的：

1. 使用 HttpClient：

       String itemIds = "abc123";
       System.out.println("itemIds are: " + itemIds + "nnn");
       String endpoint = "http://localhost:8080/api/v1/items?itemIds=" + URLEncoder.encode(itemIds, "UTF-8");
       HttpClient client = HttpClientBuilder.create().build();
       HttpGet request = new HttpGet(endpoint);
       String USER_AGENT = "Mozilla/5.0";    
       request.addHeader("User-Agent", USER_AGENT);
       request.addHeader("Content-Type", "avro/binary");
       HttpResponse response = client.execute(request);
       System.out.println("Response Code : "
           + response.getStatusLine().getStatusCode());
       byte[] bytes = EntityUtils.toByteArray(response.getEntity());
       SearchMaterializationDto deserializedReponse = SearchMaterializationAvroObjectSerializer.deserializeToSearchMaterialization(bytes);
       System.out.println(deserializedReponse.toString());
   

这种方法在执行此行时抛出java.io.EOFException SearchMaterializationDto deserializedReponse = SearchMaterializationAvroObjectSerializer.deserializeToSearchMaterialization(bytes);

这是我SearchMaterializationDto deserializedReponse = SearchMaterializationAvroObjectSerializer.deserializeToSearchMaterialization(bytes);方法：

public static SearchMaterializationDto deserializeToSearchMaterialization(byte[] buffer) {
        SearchMaterializationDto searchMaterializationDto = new SearchMaterializationDto();
        try {
            ObjectInput input = new ObjectInputStream(new ByteArrayInputStream(buffer));
            searchMaterializationDto.readExternal(input);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return searchMaterializationDto;
    }

这是我的 SearchMaterializationDto.java 类(仅列出调用的方法(：

@org.apache.avro.specific.AvroGenerated
public class SearchMaterializationDto extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
private static final org.apache.avro.io.DatumReader
    READER$ = new org.apache.avro.specific.SpecificDatumReader(SCHEMA$);
  @Override public void readExternal(java.io.ObjectInput in)
    throws java.io.IOException {
    READER$.read(this, SpecificData.getDecoder(in));
  }
}

2. 使用Avro解码器，如以下示例所示：

   private static void decoderWay(String endpoint) throws IOException {
   byte[] bytes = getBytes(endpoint);
   Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
   SpecificDatumReader<SearchMaterializationDto> reader = new SpecificDatumReader<SearchMaterializationDto>(SearchMaterializationDto.getClassSchema());
   SearchMaterializationDto searchMaterializationDto = reader.read(null, decoder);
   }
   

但它也抛出了EOFException：

Exception in thread "main" java.io.EOFException
    at org.apache.avro.io.BinaryDecoder.ensureBounds(BinaryDecoder.java:473)
    at org.apache.avro.io.BinaryDecoder.readLong(BinaryDecoder.java:160)
    at org.apache.avro.io.BinaryDecoder.doReadItemCount(BinaryDecoder.java:363)
    at org.apache.avro.io.BinaryDecoder.readMapStart(BinaryDecoder.java:408)
    at org.apache.avro.io.ValidatingDecoder.readMapStart(ValidatingDecoder.java:211)
    at org.apache.avro.generic.GenericDatumReader.readMap(GenericDatumReader.java:308)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:177)
    at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:116)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:230)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:174)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)

Answer 1

原来是我的序列化/去爬虫代码问题。我已将其更改为以下内容，并且可以正常工作：

public static byte[] serializeSearchMaterializationToByteArray(SearchMaterializationDto searchMaterializationDto) {
        return avroSerialize(SearchMaterializationDto.class, searchMaterializationDto);
    }
public static <T> byte[] avroSerialize(Class<T> clazz, Object object) {
    byte[] ret = null;
    try {
        if (object == null || !(object instanceof SpecificRecord)) {
            return null;
        }
        T record = (T) object;
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Encoder e = EncoderFactory.get().directBinaryEncoder(out, null);
        SpecificDatumWriter<T> w = new SpecificDatumWriter<T>(clazz);
        w.write(record, e);
        e.flush();
        ret = out.toByteArray();
    } catch (IOException e) {
    }
    return ret;
}
public static SearchMaterializationDto deserializeToSearchMaterialization(byte[] avroBytes) {
    return avroDeserialize(avroBytes, SearchMaterializationDto.class, SearchMaterializationDto.getClassSchema());
}
public static <T> T avroDeserialize(byte[] avroBytes, Class<T> clazz, Schema schema) {
    T ret = null;
    try {
        ByteArrayInputStream in = new ByteArrayInputStream(avroBytes);
        Decoder d = DecoderFactory.get().directBinaryDecoder(in, null);
        SpecificDatumReader<T> reader = new SpecificDatumReader<T>(clazz);
        ret = reader.read(null, d);
    } catch (IOException e) {
    }
    return ret;
}