java—将pojo转换为confluent.io中的通用记录，以便通过kafkaproducer发送

jxct1oxe 于 2021-06-07 发布在 Kafka

关注(0)|答案(1)|浏览(279)

我对Kafka和阿夫罗是完全陌生的，我试图使用合流包。我们有用于jpa的现有pojo，我希望能够简单地生成pojo的示例，而不必手动将每个值反映到通用记录中。我似乎不知道文档中是如何做到这一点的。
这些示例使用通用记录并逐个设置每个值，如下所示：

String key = "key1";
String userSchema = "{\"type\":\"record\"," +
                    "\"name\":\"myrecord\"," +
                    "\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}";
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(userSchema);
GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("f1", "value1");

record = new ProducerRecord<Object, Object>("topic1", key, avroRecord);
try {
  producer.send(record);
} catch(SerializationException e) {
  // may need to do something with it
}

有几个例子可以从一个类中获取一个模式，我找到了注解来根据需要修改这个模式。现在，我如何获取pojo的一个示例并将其按原样发送给序列化程序，并让库完成从类中匹配模式的工作，然后将值复制到泛型记录中？我做错了吗？最后我想做的是：

String key = "key1";
Schema schema = ReflectData.get().getSchema(myObject.getClass());
GenericRecord avroRecord = ReflectData.get().getRecord(myObject, schema);

record = new ProducerRecord<Object, Object>("topic1", key, avroRecord);
try {
  producer.send(record);
} catch(SerializationException e) {
  // may need to do something with it
}

谢谢！

Java avro apache-kafka kafka-producer-api

来源：https://stackoverflow.com/questions/34618695/converting-pojos-to-generic-records-in-confluent-io-to-send-through-a-kafkaprodu

1条答案

按热度按时间

brvekthn1#

我在这个示例中创建了自己的序列化程序：

public class KafkaAvroReflectionSerializer extends KafkaAvroSerializer {
   private final EncoderFactory encoderFactory = EncoderFactory.get();

   @Override
   protected byte[] serializeImpl(String subject, Object object) throws SerializationException {
      //TODO: consider caching schemas
      Schema schema = null;

      if(object == null) {
         return null;
      } else {
         try {
            schema = ReflectData.get().getSchema(object.getClass());
            int e = this.schemaRegistry.register(subject, schema);
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            out.write(0);
            out.write(ByteBuffer.allocate(4).putInt(e).array());

            BinaryEncoder encoder = encoderFactory.directBinaryEncoder(out, null);
            DatumWriter<Object> writer = new ReflectDatumWriter<>(schema);
            writer.write(object, encoder);
            encoder.flush();
            out.close();

            byte[] bytes = out.toByteArray();
            return bytes;
         } catch (IOException ioe) {
            throw new SerializationException("Error serializing Avro message", ioe);
         } catch (RestClientException rce) {
            throw new SerializationException("Error registering Avro schema: " + schema, rce);
         } catch (RuntimeException re) {
            throw new SerializationException("Error serializing Avro message", re);
         }
      }
   }
}

赞(0）回复(0）举报 2021-06-07

我来回答

java—将pojo转换为confluent.io中的通用记录，以便通过kafkaproducer发送

1条答案

相关问题

热门标签

最新问答