java - 如何在 Kafka Avro 生产者中发送对象的 ArrayList，而不为每个单独的记录调用 send 方法？

Question

这个问题已经有答案了:

Analyzing Code for Efficiency? (9 个回答)

已关闭 5 年前。

我使用下面的代码来生成 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。

Answer 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 插件，则可以为您生成这些类

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

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

将自动为 com.example.User 创建 Avro 架构 (avsc) 和 Java 类以及 getters/setters、equalsTo、toString 等。

然后，您使用 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 文件生成的类的业务逻辑，您或多或少可以将其提取到单独的实用程序类中。