找不到 key.serializer 的类

Posted 2023-03-27

【中文标题】找不到 key.serializer 的类

【英文标题】：Class for key.serializer could not be found

【发布时间】：2019-09-14 11:09:29

【问题描述】：

我正在使用 Spring Boot 应用程序向 Kafka 主题发送 JSON 数据数组，但出现以下错误：

error :org.apache.kafka.common.config.ConfigException: Invalid value 
org.apache.kafka.common.serialization.StringSerializer; for 
configuration key.serializer: Class 
org.apache.kafka.common.serialization.StringSerializer; could not be found.

我已尝试将序列化配置更改为：

props.put("key.serializer",  org.apache.kafka.common.serialization.StringSerializer;");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer;");

到

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer;");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer;");

配置文件和服务文件代码：

@Configuration
public class KafkaProducerConfig 

@Bean

private static ProducerFactory<String, String> producerConfig() 

            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer;");
            props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer;");

            // The following properties are used by LiKafkaProducerImpl
            props.put("large.message.enabled", "true");
            props.put("max.message.segment.bytes", 1000 * 1024);
            props.put("segment.serializer", DefaultSegmentSerializer.class.getName());

            props.put("auditor.class", LoggingAuditor.class.getName());

            return new DefaultKafkaProducerFactory(props);
        


@Service
public class KafkaSender 

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSender.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("$kafka.topic.name")
    private String topicName;

    public void sendData(List<Student> student) 

        System.out.println("Inside Student"+ student.toString());
        System.out.println("Inside Student"+ student);

        // TODO Auto-generated method stub
        Map<String, Object> headers = new HashMap<>();
        headers.put(KafkaHeaders.TOPIC, topicName);     

        System.out.println("\nStudent=  " + headers);

        // Construct a JSONObject from a Map.
        JSONObject HeaderObject = new JSONObject(headers);
        System.out.println("\nUsing new JSONObject() ==> " + HeaderObject);
        final String record = HeaderObject.toString();

        final int recordSize = record.length();

        kafkaTemplate.send(new GenericMessage<>(student, headers));

        LOGGER.info("Data - " + student + " sent to Kafka Topic - " + topicName);
    

POST json：

[ "studentId": "Q45678123", "名字": "abc", "姓氏": "xyz", “年龄”：“12”， “地址”： "公寓": "apt 123", “街道”：“街道信息”， “状态”：“状态”， “城市”：“城市”， “邮编”：“12345” , "studentId": "Q45678123", "名字": "abc", "姓氏": "xyz", “年龄”：“12”， “地址”： "公寓": "apt 123", “街道”：“街道信息”， “状态”：“状态”， “城市”：“城市”， “邮编”：“12345” ]

【参考方案1】：

您需要从值的末尾删除分号

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

或者你可以使用class.getName() 方法，就像你对段序列化器所做的那样，我建议这样做更安全，因为这样它可以保证你想要的序列化器在编译时可用

【讨论】：

感谢您的回答。我试过但得到以下错误：org.apache.kafka.common.errors.SerializationException: Can't convert value of class java.util.ArrayList to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer

你需要实际发送一个字符串，而不是一个列表...GenericMessage<>(student.toString()

注意：不建议发送数据列表，单个对象不会到达主题。您最终会收到一封包含整个列表的消息。如果这不是预期的，你需要一个循环

感谢您的帮助。有效。现在在消费者方面工作。但是在消费者得到以下错误-org.springframework.kafka.listener.ListenerExecutionFailedException：无法使用传入消息调用侦听器方法时，我将Deserializer值更改为String。

我建议用你的新代码创建一个新帖子，因为我已经解决了你最初的错误