如何在不为每条记录调用发送方法的情况下在 Kafka Avro 生产者中发送对象的 ArrayList？ [复制]

Posted 2023-03-06

【中文标题】如何在不为每条记录调用发送方法的情况下在 Kafka Avro 生产者中发送对象的 ArrayList？ [复制]

【英文标题】：How to send ArrayList of Objects in Kafka Avro producer without calling send method for every individual record? [duplicate]

【发布时间】：2018-04-24 20:07:54

【问题描述】：

我正在使用下面的代码将 User 类的 Avro 记录生成到 Kafka 主题中，并且工作正常；

发件人类

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Properties;

import vo.User;

public class Sender8 

    public static void main(String[] args) 

        User user = new User(10,"testName");
        Schema schema = ReflectData.get().getSchema(user.getClass());
        new GenericData.Record(schema);

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        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("schema.registry.url", "http://127.0.0.1:8081");

        KafkaProducer<String, GenericRecord> producer = new KafkaProducer<String, GenericRecord>(props);

        ReflectDatumWriter<Object> reflectDatumWriter = new ReflectDatumWriter<>(schema);
        GenericDatumReader<Object> genericRecordReader = new GenericDatumReader<>(schema);
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();

        try 
            reflectDatumWriter.write(user, EncoderFactory.get().directBinaryEncoder(bytes, null));
            GenericRecord avroRecord2 = (GenericRecord) genericRecordReader.read(null, DecoderFactory.get().binaryDecoder(bytes.toByteArray(), null));
            ProducerRecord<String, GenericRecord> record = new ProducerRecord<String, GenericRecord>("avrotesttopic1", avroRecord2);
            producer.send(record);
            producer.flush();

         catch (IOException e1) 
            e1.printStackTrace();
        

        producer.close();
    

用户类

public class User 
    int id;
    String name;

    public User(int id, String name) 
        super();
        this.id = id;
        this.name = name;
    

    public int getId() 
        return id;
    

    public void setId(int id) 
        this.id = id;
    

    public String getName() 
        return name;
    

    public void setName(String name) 
        this.name = name;
    

有时，我可能需要将对象集合作为数组列表发送，例如；

ArrayList<User> users = new ArrayList<User>();

在这种情况下，我所做的是，创建一个循环来遍历列表，选择单个记录，然后调用send() 方法，例如；

Iterator iter = users.iterator();
while (iter.hasNext()) 
   user = iter.next();
   //all other stuff here
   producer.send(record);

这很好用。但问题是，如果我的arraylist 有50 条记录，producer.send(record) 将被触发50 次。我想知道是否有任何其他更有效的方法来处理这个问题，比如对所有 50 条记录只调用一次 sender。

【问题讨论】：

编辑：更正了序列化程序类。

What does your profiler tell you? 因为除非你有证据，否则它可能不是问题。

【参考方案1】：

不清楚您的主题是期待一条包含 50 条记录的数组消息还是 50 条单独的用户消息。

如果是单个消息，这是预期的行为。没有重复调用producer.send 的开销。这就像说System.out.print，而您所做的只是将数据写入Kafka 而不是控制台。

Even see this example uses a while loop

在pom.xml 和src/main/avro 中查找Avro 插件的使用位置和LogLine 类的定义。

如果一条记录有 50 个结果，则需要为 List<User> 创建一个架构或定义一个类似的类

class UserList 
    List<User> users;

---

另外，如前文所述，如果你只使用Avro Maven Plugin，这些类都可以为你生成

For example, in AVDL 和 getting started with Avro in Java

@namespace("com.example")
protocol DomainModels 
    record User 
      int id;
      string name;
    

将为com.example.User 和getter/setter、equalsTo、toString 等自动创建Avro 模式(avsc) 和Java 类。

然后，您使用 SpecificRecord 类型而不是像这样的 GenericRecord

Producer<String, User> producer = new KafkaProducer<String, User>(props);
for (User u : list) 
    producer.send(u);

因为生成的 User 类会扩展 SpecificRecord

---

同样，如果你在 Avro 中有一个对象列表，那么 AVDL 支持数组

@namespace("com.example")
protocol DomainModels 
    record User 
      int id;
      string name;
    

    record UserList 
       array<User> users;
    

---

您目前正在做的事情的替代方法是将using an AVSC format 内联到代码中（或者更好地从文件中读取），但这基本上是ReflectDatum 正在生成的内容。

如果您只有一个没有业务逻辑的简单 Java 对象，我个人认为不需要 Reflect Avro 构建器。如果您确实需要从 AVDL/AVSC 文件生成的类的业务逻辑，您可以或多或少地将其提取到单独的实用程序类中。