Kafka生产者：发送不带模式的Avro作为字节数组。

下面是一个示例代码，展示了如何使用Kafka生产者发送不带模式的Avro作为字节数组：

import io.confluent.kafka.serializers.AbstractKafkaAvroSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class AvroProducer {

    public static void main(String[] args) {
        // 设置Kafka生产者的配置
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", KafkaAvroSerializer.class);
        properties.put("schema.registry.url", "http://localhost:8081");

        // 创建Kafka生产者
        Producer<String, byte[]> producer = new KafkaProducer<>(properties);

        // 构建Avro记录
        String topic = "test-topic";
        String avroSchema = "{\"type\":\"record\",\"name\":\"myrecord\",\"fields\":[{\"name\":\"value\",\"type\":\"string\"}]}";
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(avroSchema);
        GenericRecord avroRecord = new GenericData.Record(schema);
        avroRecord.put("value", "Hello, Kafka!");

        // 将Avro记录序列化为字节数组
        AbstractKafkaAvroSerializer avroSerializer = new KafkaAvroSerializer();
        byte[] serializedRecord = avroSerializer.serialize(topic, avroRecord);

        // 创建Kafka记录
        ProducerRecord<String, byte[]> kafkaRecord = new ProducerRecord<>(topic, serializedRecord);

        // 发送Kafka记录
        producer.send(kafkaRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("Failed to send Kafka record: " + exception.getMessage());
                } else {
                    System.out.println("Sent Kafka record to topic " + metadata.topic() +
                            ", partition " + metadata.partition() + ", offset " + metadata.offset());
                }
            }
        });

        // 关闭Kafka生产者
        producer.close();
    }
}

注意，上述示例使用了Confluent的KafkaAvroSerializer类来序列化Avro记录。确保在构建项目时包括相应的依赖，例如io.confluent:kafka-avro-serializer。

此外，根据你的环境和需求，你需要修改Kafka的地址、Avro模式、主题名称等。