Kafka骆驼Avro消费者Spring靴

Question

我一直在尝试为Spring引导Kafka骆驼Avro消费者寻找示例代码，但没有运气。我在以下URL找到了Spring Camel Kafka消费者和生产者示例：

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

我的具体问题是，一旦我的bean从Avro模式创建，并且我有了POJO类，我如何将上面的camel for spring示例告诉用户Avro序列化？具体地说，我指的是这一行代码：from（“kafka：localhost:9092？topic=test&zookeeperhost=localhost&zookeeperport=2181&groupid=group1&serializerclass=kafka.serializer.StringEncoder”）.bean（kafkaOutputBean.class）；

其中序列化程序是StringEncoder。我如何告诉骆驼使用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();
        }
    }