|
对于最新的稳定版本，请使用 spring-cloud-stream 4.3.0！ |
使用记录
在上面upppercase函数，我们将记录用作Flux<String>然后将其生成为Flux<String>. 在某些情况下，您可能需要以原始接收格式接收记录 -ReceiverRecord. 这是这样一个功能。
@Bean
public Function<Flux<ReceiverRecord<byte[], byte[]>>, Flux<String>> lowercase() {
return s -> s.map(rec -> new String(rec.value()).toLowerCase());
}请注意，在此函数中，我们将记录用作Flux<ReceiverRecord<byte[], byte[]>>然后将其生成为Flux<String>.ReceiverRecord是基本接收记录，它是专门的 KafkaConsumerRecord在 Reactor Kafka 中。当使用响应式 Kafka 绑定器时，上述函数将允许您访问ReceiverRecord类型。但是，在这种情况下，您需要为 RecordMessageConverter 提供自定义实现。默认情况下，响应式 Kafka 绑定器使用 MessagingMessageConverter，该 MessagingMessageConverter 将有效负载和标头从ConsumerRecord. 因此，当您的处理程序方法收到它时，有效负载已经从接收到的记录中提取并传递到该方法，就像我们上面看到的第一个函数一样。通过提供自定义RecordMessageConverter实现，您可以覆盖默认行为。例如，如果您想将记录作为原始记录使用Flux<ReceiverRecord<byte[], byte[]>>，则可以在应用程序中提供以下 bean 定义。
@Bean
RecordMessageConverter fullRawReceivedRecord() {
return new RecordMessageConverter() {
private final RecordMessageConverter converter = new MessagingMessageConverter();
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type payloadType) {
return MessageBuilder.withPayload(record).build();
}
@Override
public ProducerRecord<?, ?> fromMessage(Message<?> message, String defaultTopic) {
return this.converter.fromMessage(message, defaultTopic);
}
};
}然后，您需要指示框架使用此转换器进行所需的绑定。这是一个基于我们的lowercase功能。
spring.cloud.stream.kafka.bindings.lowercase-in-0.consumer.converterBeanName=fullRawReceivedRecord"lowercase-in-0是我们的lowercase功能。 对于出站（lowecase-out-0），我们仍然使用常规MessagingMessageConverter.
在toMessage实现上述，我们收到原始的ConsumerRecord (ReceiverRecord因为我们处于响应式 Binder 上下文中），然后将其包装在Message. 然后，该消息有效负载是ReceiverRecord提供给用户方法。
如果reactiveAutoCommit是false（默认），调用rec.receiverOffset().acknowledge()（或commit()） 导致偏移量被提交; 如果reactiveAutoCommit是true，助焊剂供应ConsumerRecords 代替。请参阅reactor-kafka文档和 JavaDocs 了解更多信息。