Spring Boot Kafka Camel Avro 消费者

Question

我一直在尝试寻找 Spring Boot Kafka Camel Avro 消费者的示例代码，但没有成功。我在以下网址找到了 Spring Camel Kafka 消费者和生产者示例：

https://thysmichels.com/2015/09/04/apache-camel-kafka-spring-integration/

但缺少的是 Avro 部分。我在这里查看 avro 的 Camel 文档：

http://camel.apache.org/avro.html

我的具体问题是，一旦我的 bean 是从 Avro 模式创建的并且我有 POJO 类，我如何将上面的 spring 示例中的骆驼告诉用户 Avro 序列化？具体来说，我指的是这行代码： from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181&groupId=group1&serializerClass=kafka.serializer.StringEncoder").bean(kafkaOutputBean.class);

其中的序列化器是 StringEncoder。如何告诉 Camel 使用 Avro 序列化？

Answer 1

我找到了自己的答案。因此，我想与您分享。它实际上是serializerClass=org.springframework.integration.kafka.serializer.avro.AvroSerializer。 代码非常简单，您几乎可以自己编写代码。

    public class AvroSerializer<T> {

        public T deserialize(final byte[] bytes, final DatumReader<T> reader) throws IOException {
            final Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
            return reader.read(null, decoder);
        }

        public byte[] serialize(final T input, final DatumWriter<T> writer) throws IOException {
            final ByteArrayOutputStream stream = new ByteArrayOutputStream();

            final Encoder encoder = EncoderFactory.get().binaryEncoder(stream, null);
            writer.write(input, encoder);
            encoder.flush();

            return stream.toByteArray();
        }
    }