我正在與spring-Kafka-2.1.5和一起開發(fā)微服務(wù)spring-boot-2.0.5第一個(gè)服務(wù)將向 kafka 生成一些消息，第二個(gè)服務(wù)將使用它們，而消費(fèi)我遇到了問題Caused by: java.lang.IllegalArgumentException: The class 'com.service1.model.TopicMessage' is not in the trusted packages: [java.util, java.lang, com.service2.model.ConsumeMessage]. If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).因此，從錯(cuò)誤消息來看，這是com.service1.model.TopicMessageservice1 的序列化模型。但我正在嘗試將消息反序列化為com.service2.model.ConsumeMessageservice2 中的模型并遇到此問題我在這里發(fā)現(xiàn)了同樣的問題，并嘗試了以下格式以及文檔文檔下面是我的配置 @Bean(name = "kafkaConsumerConfig") public Map<String, Object> kafkaConsumerConfig() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetconfig); props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval); props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit); }kafka消費(fèi)者工廠@Bean(name = "kafkaConsumerFactory")public ConsumerFactory<String, ConsumeMessage> kafkaConsumerFactory() { JsonDeserializer<ConsumeMessage> deserializer = new JsonDeserializer<>(); deserializer.addTrustedPackages("com.service2.model");return new DefaultKafkaConsumerFactory<String, ConsumeMessage>(kafkaConsumerConfig(),new StringDeserializer(),deserializer);}}
1 回答
喵喵時(shí)光機(jī)
TA貢獻(xiàn)1846條經(jīng)驗(yàn) 獲得超7個(gè)贊
您需要header precedence在反序列化器中禁用：
/**
* Construct an instance with the provided target type, and
* useHeadersIfPresent with a default {@link ObjectMapper}.
* @param targetType the target type.
* @param useHeadersIfPresent true to use headers if present and fall back to target
* type if not.
* @since 2.2
*/
public JsonDeserializer(Class<? super T> targetType, boolean useHeadersIfPresent) {
useHeadersIfPresent參數(shù)必須配置false為. 這樣inferred，將使用一種類型，并且將忽略標(biāo)頭值。
如果你不使用，你應(yīng)該考慮用類似的邏輯spring-kafka-2.2實(shí)現(xiàn)你自己的： https ://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/ springframework/kafka/support/serializer/JsonDeserializer.javaJsonDeserializer
添加回答
舉報(bào)
0/150
提交
取消