目前与spring-kafka 2.1.8.RELEASE一起使用spring-boot 2.0.4。我想简化交换,将对象发送到kafka模板,并使用json作为格式。但是,一些需要反序列化的消息包含java.time.LocalDateTime。所以我的设置是
配置(application.yml):
spring: jackson: serialization: write_dates_as_timestamps: false kafka: consumer: group-id: foo enable-auto-commit: true key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: my.package producer: value-serializer: org.springframework.kafka.support.serializer.JsonSerializer key-serializer: org.apache.kafka.common.serialization.StringSerializer properties: spring.json.trusted.packages: my.package retries: 3 acks: all
至于应该工作的杰克逊依赖关系,我的依赖关系树是:
[INFO] | | +- com.fasterxml.jackson.core:jackson-databind:jar:2.9.6:compile [INFO] | | | +- com.fasterxml.jackson.core:jackson-annotations:jar:2.9.0:compile [INFO] | | | \- com.fasterxml.jackson.core:jackson-core:jar:2.9.6:compile [INFO] | | \- com.fasterxml.jackson.datatype:jackson-datatype-jsr310:jar:2.9.6:compile [INFO] | | +- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:jar:2.9.6:compile [INFO] | | \- com.fasterxml.jackson.module:jackson-module-parameter-names:jar:2.9.6:compile
但是,这会产生以下错误:
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition Foo-0 at offset 4. If needed, please seek past the record to continue consumption. Caused by: org.apache.kafka.common.errors.SerializationException: Can't deserialize data [[123, 34, 105, 100, 34, 58, 34, 97, 50, 99, 50, 56, 99, 99, 101, 97, 49, 98, 98, 52, 51, 97, 97, 56, 53, 50, 49, 53, 99, 101, 49, 54, 57, 48, 52, 51, 51, 98, 51, 45, 50, 34, 44, 34, 97, 117, 116, 104, 111, 114, 34, 58, 34, 97, 110, 116, 111, 110, 105, 111, 34, 44, 34, 99, 114, 101, 97, 116, 101, 100, 34, 58, 123, 34, 104, 111, 117, 114, 34, 58, 49, 56, 44, 34, 109, 105, 110, 117, 116, 101, 34, 58, 52, 48, 44, 34, 115, 101, 99, 111, 110, 100, 34, 58, 53, 49, 44, 34, 110, 97, 110, 111, 34, 58, 51, 50, 53, 48, 48, 48, 48, 48, 48, 44, 34, 100, 97, 121, 79, 102, 89, 101, 97, 114, 34, 58, 50, 52, 48, 44, 34, 100, 97, 121, 79, 102, 87, 101, 101, 107, 34, 58, 34, 84, 85, 69, 83, 68, 65, 89, 34, 44, 34, 109, 111, 110, 116, 104, 34, 58, 34, 65, 85, 71, 85, 83, 84, 34, 44, 34, 100, 97, 121, 79, 102, 77, 111, 110, 116, 104, 34, 58, 50, 56, 44, 34, 121, 101, 97, 114, 34, 58, 50, 48, 49, 56, 44, 34, 109, 111, 110, 116, 104, 86, 97, 108, 117, 101, 34, 58, 56, 44, 34, 99, 104, 114, 111, 110, 111, 108, 111, 103, 121, 34, 58, 123, 34, 99, 97, 108, 101, 110, 100, 97, 114, 84, 121, 112, 101, 34, 58, 34, 105, 115, 111, 56, 54, 48, 49, 34, 44, 34, 105, 100, 34, 58, 34, 73, 83, 79, 34, 125, 125, 44, 34, 97, 103, 103, 114, 101, 103, 97, 116, 101, 73, 100, 34, 58, 34, 97, 50, 99, 50, 56, 99, 99, 101, 97, 49, 98, 98, 52, 51, 97, 97, 56, 53, 50, 49, 53, 99, 101, 49, 54, 57, 48, 52, 51, 51, 98, 51, 34, 44, 34, 118, 101, 114, 115, 105, 111, 110, 34, 58, 48, 44, 34, 112, 114, 105, 122, 101, 73, 110, 102, 111, 34, 58, 123, 34, 110, 117, 109, 98, 101, 114, 79, 102, 87, 105, 110, 110, 101, 114, 115, 34, 58, 49, 44, 34, 112, 114, 105, 122, 101, 80, 111, 111, 108, 34, 58, 49, 48, 44, 34, 112, 114, 105, 122, 101, 84, 97, 98, 108, 101, 34, 58, 91, 49, 48, 93, 125, 125]] from topic [Foo] Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Expected array or string. at [Source: (byte[])"{"id":"a2c28ccea1bb43aa85215ce1690433b3-2","author":"foo","created":{"hour":18,"minute":40,"second":51,"nano":325000000,"dayOfYear":240,"dayOfWeek":"TUESDAY","month":"AUGUST","dayOfMonth":28,"year":2018,"monthValue":8,"chronology":{"calendarType":"iso8601","id":"ISO"}},"aggregateId":"a2c28ccea1bb43aa85215ce1690433b3","version":0,"prizeInfo":{"numberOfWinners":1,"prizePool":10,"prizeTable":[10]}}"; line: 1, column: 73] (through reference chain: my.package.Foo["created"]) at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:63) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.DeserializationContext.reportInputMismatch(DeserializationContext.java:1342) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.DeserializationContext.handleUnexpectedToken(DeserializationContext.java:1138) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.datatype.jsr310.deser.JSR310DeserializerBase._handleUnexpectedToken(JSR310DeserializerBase.java:99) ~[jackson-datatype-jsr310-2.9.6.jar:2.9.6] at com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer.deserialize(LocalDateTimeDeserializer.java:141) ~[jackson-datatype-jsr310-2.9.6.jar:2.9.6] at com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer.deserialize(LocalDateTimeDeserializer.java:39) ~[jackson-datatype-jsr310-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.deser.impl.FieldProperty.deserializeAndSet(FieldProperty.java:136) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:369) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:159) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.ObjectReader._bindAndClose(ObjectReader.java:1611) ~[jackson-databind-2.9.6.jar:2.9.6] at com.fasterxml.jackson.databind.ObjectReader.readValue(ObjectReader.java:1234) ~[jackson-databind-2.9.6.jar:2.9.6] at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:228) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE] at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:923) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.internals.Fetcher.access$2600(Fetcher.java:93) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1100) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1200(Fetcher.java:949) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:570) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:531) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1154) ~[kafka-clients-1.0.2.jar:na] at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111) ~[kafka-clients-1.0.2.jar:na] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:699) ~[spring-kafka-2.1.8.RELEASE.jar:2.1.8.RELEASE] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_131] at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_131] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]
因此,我尝试了以下操作,但到目前为止还没有成功:1.Custom ObjectMapper声明为bean
@Bean public ObjectMapper objectMapper() { ObjectMapper objectMapper = new ObjectMapper(); objectMapper.registerModule(new JavaTimeModule()); objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); return objectMapper; }
2. LocalDateTime字段上的Serializer批注
为了确保我具有正确的对象映射器设置和必要的依赖关系,我创建了一个rest控制器,将响应模拟为json,作为rest终结点返回带有日期时间字段的对象,这将正确返回;样品:
[ { "playerId": "foo", "points": 10, "entryDateTime": "2018-08-19T09:30:20.051" }, { "playerId": "bar", "points": 3, "entryDateTime": "2018-08-27T09:30:20.051" } ]
Farhad Abdol.. 5
将Json(De)Serializer构造函数与对象映射器参数一起使用对我来说很有效。我在对具有java.time.Instant字段的pojo进行反序列化时遇到了麻烦,因此,在对同一个org.apache.kafka.common.errors.SerializationException
*** 进行数小时的故障排除后,我终于意识到了(借助此处的答案)不是春天,而是卡夫卡自己的序列化。给定我拥有的objectmapper bean,我通过将其自动装配到我的kafka生产者和消费者设置的JsonSerializer
和JsonDeserializer
参数中来解决。
@Configuration
public class JacksonConfig {
@Bean
@Primary
public ObjectMapper objectMapper(Jackson2ObjectMapperBuilder builder) {
ObjectMapper objectMapper = builder.build();
objectMapper.registerModule(new JavaTimeModule());
objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
return objectMapper;
}
}
@Configuration
public class KafkaProducerConfig {
@Value(value="${kafka.bootstrapAddress}")
private String bootstrapAddress;
@Autowired
private ObjectMapper objectMapper;
@Bean
public KafkaTemplate orderKafkaTemplate(){
Map props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props, new StringSerializer(), new JsonSerializer(objectMapper));
return new KafkaTemplate<>(producerFactory);
}
}
@Configuration
public class KafkaConsumerConfig {
@Value(value="${kafka.bootstrapAddress}")
private String bootstrapAddress;
@Value(value="${kafka.consumer.groupId}")
private String groupId;
@Autowired
private ObjectMapper objectMapper;
@Bean
public ConcurrentKafkaListenerContainerFactory orderKafkaListenerContainerFactory(){
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
Map props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
ConsumerFactory cOnsumerFactory= new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(Order.class, objectMapper));
factory.setConsumerFactory(consumerFactory);
return factory;
}
}
(为进一步清晰起见,显示了Pojo)
public class Order {
private long accountId;
private long assetId;
private long quantity;
private long price;
private Instant createdOn = Instant.now();
// no args constructor, constructor with params for all fields except createdOn, and getters/setters for all fields omitted
***通常的原因是: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of 'java.time.Instant' (no Creators, like default construct, exist): cannot deserialize from object value (no delegate- or property-based Creator) at [Source: (byte[])"{"accountId":1,"assetId":2,"quantity":100,"price":1000,"createdOn":{"epochSecond":1558570217,"nano":728000000}}"...
将Json(De)Serializer构造函数与对象映射器参数一起使用对我来说很有效。我在对具有java.time.Instant字段的pojo进行反序列化时遇到了麻烦,因此,在对同一个org.apache.kafka.common.errors.SerializationException
*** 进行数小时的故障排除后,我终于意识到了(借助此处的答案)不是春天,而是卡夫卡自己的序列化。给定我拥有的objectmapper bean,我通过将其自动装配到我的kafka生产者和消费者设置的JsonSerializer
和JsonDeserializer
参数中来解决。
@Configuration
public class JacksonConfig {
@Bean
@Primary
public ObjectMapper objectMapper(Jackson2ObjectMapperBuilder builder) {
ObjectMapper objectMapper = builder.build();
objectMapper.registerModule(new JavaTimeModule());
objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
return objectMapper;
}
}
@Configuration
public class KafkaProducerConfig {
@Value(value="${kafka.bootstrapAddress}")
private String bootstrapAddress;
@Autowired
private ObjectMapper objectMapper;
@Bean
public KafkaTemplate orderKafkaTemplate(){
Map props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props, new StringSerializer(), new JsonSerializer(objectMapper));
return new KafkaTemplate<>(producerFactory);
}
}
@Configuration
public class KafkaConsumerConfig {
@Value(value="${kafka.bootstrapAddress}")
private String bootstrapAddress;
@Value(value="${kafka.consumer.groupId}")
private String groupId;
@Autowired
private ObjectMapper objectMapper;
@Bean
public ConcurrentKafkaListenerContainerFactory orderKafkaListenerContainerFactory(){
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
Map props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
ConsumerFactory cOnsumerFactory= new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(Order.class, objectMapper));
factory.setConsumerFactory(consumerFactory);
return factory;
}
}
(为进一步清晰起见,显示了Pojo)
public class Order {
private long accountId;
private long assetId;
private long quantity;
private long price;
private Instant createdOn = Instant.now();
// no args constructor, constructor with params for all fields except createdOn, and getters/setters for all fields omitted
***通常的原因是: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of 'java.time.Instant' (no Creators, like default construct, exist): cannot deserialize from object value (no delegate- or property-based Creator) at [Source: (byte[])"{"accountId":1,"assetId":2,"quantity":100,"price":1000,"createdOn":{"epochSecond":1558570217,"nano":728000000}}"...