我正在阅读Kafka主题,其中包含使用KafkaAvroEncoder序列化的Avro消息(它会自动使用主题注册模式).我正在使用maven-avro-plugin生成普通的Java类,我想在阅读时使用它.

KafkaAvroDecoder仅支持反序列化为GenericData.Record类型,(在我看来)这些类型错过了使用静态类型语言的全部要点.我的反序列化代码目前看起来像这样：

    SpecificDatumReader<event> reader = new SpecificDatumReader<>(
        event.getClassSchema() // event is my class generated from the schema
    );
    byte[] in = ...; // my input bytes;
    ByteBuffer stuff = ByteBuffer.wrap(in);
    // the KafkaAvroEncoder puts a magic byte and the ID of the schema (as stored 
    //   in the schema-registry) before the serialized message
    if (stuff.get() != 0x0) {
        return;
    }
    int id = stuff.getInt();

    // lets just ignore those special bytes
    int length = stuff.limit() - 4 - 1;
    int start = stuff.position() + stuff.arrayOffset();

    Decoder decoder = DecoderFactory.get().binaryDecoder(
        stuff.array(), start, length, null
    );
    try {
        event ev = reader.read(null, decoder);
    } catch (IOException e) {
        e.printStackTrace();
    }

我发现我的解决方案很麻烦,所以我想知道是否有更简单的解决方案来做到这一点.

解决方法:

由于评论,我能够找到答案.秘诀是使用指定使用特定Avro阅读器的属性来实例化KafkaAvroDecoder,即：

    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "...");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_C‌ONFIG, "...");
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
    VerifiableProp vProps = new VerifiableProperties(props);

    KafkaAvroDecoder decoder = new KafkaAvroDecoder(vProps);
    MyLittleData data = (MyLittleData) decoder.fromBytes(input);

相同的配置适用于直接使用KafkaConsumer< K,V>的情况. class(我使用Storm-kafka项目中的KafkaSpout使用SimpleConsumer从Storm中使用KafkaSpout,所以我必须手动反序列化消息.勇敢的是有storm-kafka-client项目,这样做通过使用新的样式消费者自动).