|
此版本仍在开发中，尚不被认为是稳定的。对于最新的稳定版本，请使用 spring-cloud-stream 4.3.0！ |
时间戳提取器
Kafka Streams 允许您根据各种时间戳概念控制使用者记录的处理。
默认情况下，Kafka Streams 会提取嵌入在使用者记录中的时间戳元数据。
您可以通过提供不同的TimestampExtractor每个输入绑定的实现。
以下是有关如何做到这一点的一些详细信息。
@Bean
public Function<KStream<Long, Order>,
Function<KTable<Long, Customer>,
Function<GlobalKTable<Long, Product>, KStream<Long, Order>>>> process() {
return orderStream ->
customers ->
products -> orderStream;
}
@Bean
public TimestampExtractor timestampExtractor() {
return new WallclockTimestampExtractor();
}然后你设置上面的TimestampExtractor每个消费者绑定的 Bean 名称。
spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.timestampExtractorBeanName=timestampExtractor
spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.timestampExtractorBeanName=timestampExtractor
spring.cloud.stream.kafka.streams.bindings.process-in-2.consumer.timestampExtractorBeanName=timestampExtractor"如果跳过输入使用者绑定来设置自定义时间戳提取器，则该使用者将使用默认设置。