java - 响应式(Reactive)@StreamListener

Question

只是一个快速的。

是否可以使用 Spring Cloud Stream 进行响应式(Reactive) @StreamListener Input？我的意思是这样的:

@StreamListener
public void log(@Input(Sink.INPUT) Flux<String> strings)
{
    strings.log();
}

在文档中我只看到了这样的例子:

@StreamListener(Sink.INPUT)
public void log(String message)
{
    log.info(message);
}

或者像这样的处理器:

@StreamListener
@Output(Source.OUTPUT)
public Flux<String> log(@Input(Sink.INPUT) Flux<String> strings)
{
    return strings.map(String::toUpperCase);
}

当我运行第一个片段中的代码时，我得到了一个Dispatcher has no subscribers异常。

Answer 1

前两个是等价的。如果遇到错误，请打开 GitHub 问题。

编辑:

这两个注解是等价的，但是如果对输入通量有订阅，监听器只会订阅输入。为了完全等效，响应式(Reactive)应该是:

@StreamListener
public void log(@Input(Sink.INPUT) Flux<String> strings)
{
    strings.log().subscribe();
}

或

@StreamListener
public void log(@Input(Sink.INPUT) Flux<String> strings)
{
    strings.subscribe(log::info);
}

只需在 flux 上调用 log() 即可创建另一个 Flux 而无需自己订阅。