我是Kafka和阿夫罗的傀儡。所以我一直在努力让生产者/消费者运作起来。到目前为止，我已经能够使用以下方法生成和使用简单的字节和字符串：producer的配置：

Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

    Schema.Parser parser = new Schema.Parser();
    Schema schema = parser.parse(USER_SCHEMA);
    Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);

    KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);

    for (int i = 0; i < 1000; i++) {
        GenericData.Record avroRecord = new GenericData.Record(schema);
        avroRecord.put("str1", "Str 1-" + i);
        avroRecord.put("str2", "Str 2-" + i);
        avroRecord.put("int1", i);

        byte[] bytes = recordInjection.apply(avroRecord);

        ProducerRecord<String, byte[]> record = new ProducerRecord<>("mytopic", bytes);
        producer.send(record);
        Thread.sleep(250);
    }
    producer.close();
}

现在一切都很好，当我尝试序列化pojo时，问题来了。因此，我能够使用avro提供的实用程序从pojo获取avroschema。对架构进行硬编码，然后尝试创建一个通用记录，以通过kafkaproducer发送，现在将生产者设置为：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.KafkaAvroSerializer");

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(USER_SCHEMA); // this is the Generated AvroSchema
KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);

这就是问题所在：在我使用kafkaavroserializer的那一刻，由于缺少必需的参数schema.registry.url，生产者没有出现
我仔细阅读了为什么需要这样做，这样我的消费者就能够破译制作人发给我的任何东西。但是这个模式不是已经嵌入到avromessage中了吗？如果有人能与kafkaavroserializer共享一个使用kafkaproducer的工作示例，而不必指定schema.registry.url，那就太好了
也非常感谢您提供有关模式注册表实用程序的任何见解/资源。
谢谢！