spring-boot - 使用 Spring Webflux 时在后台运行 Mono，同时返回响应

Question

此问题与 Return immediately in spring web flux 有关但我认为不一样(至少那里的答案对我来说并不令人满意)。

我有一个函数返回 Mono调用时会启动一个长时间运行的作业。当调用 Spring Webflux HTTP API 时调用此函数。这是一个例子:

@PutMapping("/{jobId}")
fun startNewJob(@PathVariable("jobId") jobId: String,
                request: ServerHttpRequest): Mono<ResponseEntity<Unit>> {
    val longRunningJob : Mono<Job> = startNewJob(jobId)
    longRunningJob.map { job ->
        val jobUri = generateJobUri(request, job.id)
        ResponseEntity.created(jobURI).build<Unit>()
    }
}

上面代码的问题是“201 Created”是在长时间运行的作业完成后创建的。我想启动 longRunningJob在后台并立即返回“201 Created”。

我也许可以做这样的事情:

@PutMapping("/{jobId}")
fun startNewJob(@PathVariable("jobId") jobId: String,
                request: ServerHttpRequest): Mono<ResponseEntity<Unit>> {

    startNewJob(jobId)
        .subscribeOn(Schedulers.newSingle("thread"))
        .subscribe()

    val jobUri = generateJobUri(request, job.id)
    val response = ResponseEntity.created(jobURI).build<Unit>()
    Mono.just(response)
}

但对我来说，必须调用 subscribe() 似乎不太习惯。手动(例如 intellij 提示我在非阻塞范围内调用 subscribe())。不使用显式 subscribe 组合两个“流”是否有更好的方法？ ?如果是这样，我该如何修改 startNewJob上面的函数能实现这个吗？

Answer 1

AFAIK，使用 subscribe 之一方法是在后台真正启动具有自己生命周期的作业的唯一方法(与返回的发布者无关)。

如果您要使用其中一个运算符来组合作业发布者和响应发布者(例如 zip 或 merge )，那么作业发布者的生命周期将与响应发布者相关联，这不是您想要的对于后台工作。

您可能要考虑的一件事是在响应发布者流中启动后台作业，而不是直接在方法主体中。例如通过 doOnSubscibe或来自响应上游的运算符(operator)。

这会将后台作业的开始与响应发布者的 onSubscribe 事件联系起来，但仍允许它在后台完成。

另请注意，如果您希望能够取消后台作业(例如，可能在应用程序关闭期间)，您需要保存 Disposable从 subscribe 返回这样您以后可以调用dispose在上面。这可能最好通过某种类型的 BackgroundJobManager 来完成，它可以跟踪所有正在运行的作业。