将带有参数的案例类转换为Avro消息以发送给Kafka

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

public class AvroProducer {
    private static final String SCHEMA_JSON = "{\"type\":\"record\",\"name\":\"Example\",\"fields\":[{\"name\":\"param1\",\"type\":\"string\"},{\"name\":\"param2\",\"type\":\"int\"}]}";
    private static final String KAFKA_TOPIC = "example_topic";
    private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:9092";

    public static void main(String[] args) {
        // 创建Avro Schema
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(SCHEMA_JSON);

        // 创建Kafka Producer
        KafkaProducer<String, byte[]> producer = new KafkaProducer<>(getKafkaConfig());

        // 创建案例类实例
        Example example = new Example("value1", 123);

        // 将案例类转换为Avro消息
        GenericRecord record = new GenericData.Record(schema);
        record.put("param1", example.getParam1());
        record.put("param2", example.getParam2());

        byte[] avroMessage = serializeAvroMessage(record, schema);

        // 发送Avro消息到Kafka
        ProducerRecord<String, byte[]> kafkaRecord = new ProducerRecord<>(KAFKA_TOPIC, avroMessage);
        producer.send(kafkaRecord);

        producer.close();
    }

    private static byte[] serializeAvroMessage(GenericRecord record, Schema schema) {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);

        try {
            org.apache.avro.io.DatumWriter<GenericRecord> writer = new org.apache.avro.specific.SpecificDatumWriter<>(schema);
            writer.write(record, encoder);
            encoder.flush();
            outputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return outputStream.toByteArray();
    }

    private static Map<String, Object> getKafkaConfig() {
        Map<String, Object> config = new HashMap<>();
        config.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
        config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        config.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        return config;
    }
}

class Example {
    private String param1;
    private int param2;

    public Example(String param1, int param2) {
        this.param1 = param1;
        this.param2 = param2;
    }

    public String getParam1() {
        return param1;
    }

    public int getParam2() {
        return param2;
    }
}