使用 avro 架构将 kafka 使用者消息从字符串解码为 avro

2 回答

• 1

  按照以下步骤 -

  1.Create avro 架构中的对象

  java -jar /path/to/avro-tools-1.8.2.jar compile schema <schema file> <destination>
  
  eg.
  java -jar /path/to/avro-tools-1.8.2.jar compile schema user.avsc .
  

  这将基于模式的命名空间在包中生成适当的源文件

  • 使用 avro 架构和上面生成的类反序列化例如，如果上面的命令将源类创建为 User.class，则反序列化数据，如下所示

  private static void deserialize() {
          try {
              // Deserialize Users from disk
              DatumReader<User> userDatumReader = new SpecificDatumReader<>(User.class);
              DataFileReader<User> dataFileReader = new DataFileReader<User>(new File("users.avro"), userDatumReader);
              User user = null;
              while (dataFileReader.hasNext()) {
                  // Reuse user object by passing it to next(). This saves us from
                  // allocating and garbage collecting many objects for files with
                  // many items.
                  user = dataFileReader.next(user);
                  System.out.println("deserialized : "+user);
              }
          } catch (IOException e) {
              e.printStackTrace();
          }
      }
  

  回复于 2024-05-03T15:19:49+08:00
• 0

  以下是我的解决方法：此代码将以 Avro 架构格式打印消费者消息。

  props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
   props.put("value.deserializer","io.confluent.kafka.serializers.KafkaAvroDeserializer");
   props.put("schema.registry.url", "http://kafka-.XXX");
  
   KafkaConsumer<String, avro_schema> consumer = new KafkaConsumer<>(props);
  
   consumer.subscribe(Arrays.asList("topic"));
   //infinite poll loop
  
   try {
        while (true) {
  
             ConsumerRecords<String, avro_schema> records = consumer.poll(200L);
             for (ConsumerRecord<String, avro_schema> record : records) {
                     System.out.println(record.value());
  
             }
        }
   }
  

  回复于 2024-05-03T15:19:49+08:00