代码之家 › 专栏 › 技术社区 › KJQ

如何将2个Spring Mongo反应式保存操作链接在一起?

spring-mongodb project-reactor spring-webflux

KJQ · 技术社区 · 6 年前

我有一个测试用例,试图从两个不同的服务启动两个save操作。两者都返回Mono(我没有使用Spring反应存储库)。

我想按顺序执行以下操作:

保存配置文件
创建特征并使用配置文件更新
转换/保存特征(返回特征)

他们单独工作。当我尝试将这两个链接在一起时,第二个操作是否挂起取决于我所做的(或只是不启动)。

我想它是链接/订阅到第一个的?

Profile profile = new GenericProfile();

Object o = profileService.saveProfile(profile)
    .log()
    .flatMap(pp -> {
        TrackingTrait trait = new TrackingTrait(
            "cid",
            "tid",
            pp.getId(),
            null,
            "h",
            "p",
            null);
        return Mono
            .just(trait)
            .log();
    })
    .doOnNext(n -> log.debug("1 {}", n.getProfileId()))
    .flatMap(tt -> this.trackingService
        .track(tt)
        .log())
    .doOnNext(n -> log.debug("2 {}", n.getProfileId()))
    .block();

我的输出如下所示,打印第二条日志语句时从未完成(只是旋转)。

[ INFO] reactor.Mono.FlatMap.1                   : | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
[ INFO] reactor.Mono.FlatMap.1                   : | request(unbounded)
[ INFO] reactor.Mono.FlatMap.1                   : | onNext(io.logicdrop.profiles.services.GenericProfile@1a01ffff)
[ INFO] reactor.Mono.Just.2                      : | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
[ INFO] reactor.Mono.Just.2                      : | request(unbounded)
[ INFO] reactor.Mono.Just.2                      : | onNext(tid)
[DEBUG] i.l.analytics.AnalyticsPersistTest       : 1 5ad9ed16a29e0e2f82775a82
[ INFO] reactor.Mono.FlatMap.3                   : | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
[ INFO] reactor.Mono.FlatMap.3                   : | request(unbounded)

1 回复 | 直到 6 年前

KJQ 6 年前

天哪,这看起来是Spring/Mongo处理索引和反应性内容的方式中的一个bug。第二次尝试创建索引时,它挂起了。

我偶然发现了这个 Reactive mongo hangs with inheritance hierarchies defined in different gradle project

如果我用这个 解决方法 并删除所有索引注释(@index和@compoundex),至少在第二个DAO中是这样的。

现在,我的日志输出如下所示:

[ INFO] reactor.Mono.FlatMap.1                   : | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
[ INFO] reactor.Mono.FlatMap.1                   : | request(unbounded)
[ INFO] reactor.Mono.FlatMap.1                   : | onNext(io.logicdrop.profiles.services.GenericProfile@28519bfb)
[ INFO] reactor.Mono.Just.2                      : | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
[ INFO] reactor.Mono.Just.2                      : | request(unbounded)
[ INFO] reactor.Mono.Just.2                      : | onNext(tid)
[DEBUG] i.l.analytics.AnalyticsPersistTest       : 1 5ad9f248a29e0e307f885e5e
[ INFO] reactor.Mono.FlatMap.3                   : | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
[ INFO] reactor.Mono.FlatMap.3                   : | request(unbounded)
[ INFO] reactor.Mono.Just.2                      : | onComplete()
[ INFO] reactor.Mono.FlatMap.1                   : | onComplete()
[ INFO] reactor.Mono.FlatMap.3                   : | onNext(tid)
[DEBUG] i.l.analytics.AnalyticsPersistTest       : 2 5ad9f248a29e0e307f885e5e
[ INFO] reactor.Mono.FlatMap.3                   : | onComplete()