我在做微服务 spring-Kafka-2.1.5
以及 spring-boot-2.0.5
第一个服务将产生一些消息给Kafka和第二个将消费他们,而消费我有问题
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 (*).
所以从错误信息来看这是 com.service1.model.TopicMessage
service1的序列化模型。但我正在尝试将消息反序列化到模型中 com.service2.model.ConsumeMessage
它在服务中2并且有这个问题
我在这里发现了同样的问题,并尝试了下面的格式和文档
下面是我的配置
@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消费工厂
@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);
}
KafkalistenerContainer工厂
@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;
}
1条答案
按热度按时间xkftehaa1#
你需要禁用
header precedence
在反序列化程序中:这个
useHeadersIfPresent
参数必须配置为false
. 那样的话inferred
类型将被使用,而标头值将被忽略。如果你不使用
spring-kafka-2.2
,你应该考虑实施你自己的JsonDeserializer
逻辑相似:https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/jsondeserializer.java