我使用下面的代码来生成 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。
最佳答案
不清楚您的主题是否需要一条包含 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 文件生成的类的业务逻辑,您或多或少可以将其提取到单独的实用程序类中。
关于java - 如何在 Kafka Avro 生产者中发送对象的 ArrayList,而不为每个单独的记录调用 send 方法?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50009999/