将 Spring Cloud Stream 转换为使用响应式云功能

Question

目前我有类似这样的 Spring Boot 应用程序。

@Component
@EnableBinding(Source::class)
class EmailMessageProducer(private val source: Source) {

    suspend fun send(textMessage: TextMessage) {

        source.output().send(
            MessageBuilder.withPayload(textMessage).setHeader("service", "test").build()
        )
    }
}

我想在这里使用反应模式使用 Spring Cloud Function。

此外，我当前的解决方案是非阻塞的吗？我问这个是因为这是我第一次在这种情况下使用 Kotlin 协程。

Java 解决方案也适用于我，因为我只是想理解这里的概念。

Answer 1

您正在寻找的是响应式供应商（例如，Supplier<Flux>）。

在你的情况下，它看起来像这样：

@SpringBootApplication
public class SomeApplication {

    @Bean
    public Supplier<Flux<Message<String>>> messageProducer() {
        return () -> Flux.just(MessageBuilder.withPayload(textMessage).setHeader("service", "test").build());
    }
}

提供spring.cloud.function.definition=messageProducer 属性，就差不多了。

显然，上面的示例生成了一个包含单个项目的有限流，但可以随意修改返回的通量。事实上，我们会更详细地讨论这个问题here。