本文介紹了Spring Kafka模型不在受信任的包中的處理方法，對(duì)大家解決問(wèn)題具有一定的參考價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧！
問(wèn)題描述
我正在使用spring-Kafka-2.1.5和spring-boot-2.0.5處理微服務(wù)
第一個(gè)服務(wù)將向Kafka生成一些消息，第二個(gè)服務(wù)將消費(fèi)這些消息，而消費(fèi)時(shí)我遇到了問(wèn)題
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ò)誤消息來(lái)看，這是Service1的com.service1.model.TopicMessage序列化模型。但我正在嘗試將消息反序列化到Service2中的模型com.service2.model.ConsumeMessage中，并出現(xiàn)此問(wèn)題
我發(fā)現(xiàn)了同樣的問(wèn)題here，并嘗試了下面的格式和文檔docs
下面是我的配置
@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);
}
kafkaConsumer erFactory
@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);
}
kafkaListenerContainerFactory
@Bean(name = "kafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, ConsumeMessage > kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, ConsumeMessage > factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConcurrency(Integer.parseInt(threads));
factory.setBatchListener(true);
factory.setConsumerFactory(kafkaConsumerFactory());
factory.getContainerProperties().setPollTimeout(Long.parseLong(pollTimeout));
factory.getContainerProperties().setAckMode(AckMode.BATCH);
return factory;
}
推薦答案
您需要禁用反序列化程序中的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)頭值。
如果您不使用spring-kafka-2.2，則應(yīng)考慮使用類似的邏輯實(shí)現(xiàn)您自己的JsonDeserializer：https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java
這篇關(guān)于Spring Kafka模型不在受信任的包中的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，
