Mono与Spring Cloud Stream

Question

我在尝试将Reactor的Mono与Spring Cloud Stream一起使用时遇到了一个问题，无法真正弄清楚发生了什么。

想象一下，我有一个这样的听众：

@StreamListener
@Output(Urls.OUTUT)
public Flux<String> expandUrls(@Input(Urls.INPUT) Flux<String> urlFormats)
{
    return urlFormats
        .map(this::expandUrl)
        .flatMapIterable(urls -> urls);
}

因此，它基本上是将一个格式如下的url扩展成这样

< code > http://www . example . com/page/1

http://www.example.com/page/2

http://www.example.com/page/3

它按预期工作，但是当我尝试这样做时：

@StreamListener
@Output(Urls.OUTPUT)
public Flux<String> expandUrls(@Input(Urls.INPUT) Mono<String> urlFormats)
{
    return urlFormats
        .repeat(3)
        .zipWith(pageNumbers)
        .map(this::formatUrl);
}

其中pageNumber是Flux.fromStream（Stream.eiterate（1，p-

我得到以下例外

Caused by: java.lang.IllegalArgumentException: A method annotated with @StreamListener may use @Input or @Output annotations only in declarative mode and for parameters that are binding targets or convertible from binding targets.

我通过这样做来消除这个异常

@StreamListener(value = Urls.INPUT)
@Output(Urls.OUTPUT)
public Flux<String> expandUrls(Mono<String> urlFormats)
{
    return urlFormats
        .repeat(3)
        .zipWith(pageNumbers)
        .map(this::formatUrl);
}

但现在我明白了：

Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'http': was expecting ('true', 'false' or 'null')
 at [Source: http://www.example.com/page-%d,1,0.html;

我的问题是:如何使用Mono和Spring Cloud Stream。有可能这样使用它吗？如果是，那么怎么做呢？哦，我用Kafka做中间人。




Answer 1

Spring Cloud Stream StreamListener的@输入参数类型仅支持反应器类型Flux，因为它比在@输入参数类型上具有Mono更适合反应性流应用程序。