Flux doOnComplete 无参数问题：可获取上游返回值的替代方案有哪些？

doOnComplete是流完成信号的回调算子，仅负责触发流结束的通知逻辑，本身不携带上游发射的元素值，因此无法直接在该算子中获取目标响应对象。你可以通过以下几种方案实现需求：

• 方案一：使用last()算子捕获最终完成响应（最优方案）
  因为takeUntil算子的执行逻辑是：持续发射上游元素，直到断言返回true，此时会先把触发断言的元素（也就是状态为completed的响应）发射出来，再触发流完成信号。因此你可以通过last()提取流的最后一个元素，直接在doOnNext中处理即可，修改后代码如下：

private void status(String id) {
        restClient
                .get(url + "/status/" + id)
                .send(StatusResponse.class)
                .repeatWhen(completed -> completed.delayElements(Duration.ofMillis(2000)))
                .takeUntil(statusResponse->statusResponse.getStatus().equals("completed"))
                .last() // 提取流的最后一个元素，即状态为completed的响应
                .doOnNext(completedResponse -> {
                    // 此处直接编写你的后续计算逻辑
                    log.info("获取到完成响应：{}", completedResponse);
                })
                .doOnError(e -> log.error("Error :: " + e.getMessage(), e))
                .subscribe();
    }

本方案无需重复判断响应状态，代码简洁无额外性能损耗，是适配你当前业务场景的最优选择。如果担心极端情况下流为空引发NoSuchElementException，可以替换为lastOrDefault(默认值)或者takeLast(1).singleOptional()做兼容处理。

• 方案二：在doOnNext中做状态判断
  如果你需要同时处理中间轮询得到的非完成状态响应，可以直接在takeUntil之前添加doOnNext算子，通过状态判断处理完成响应：

private void status(String id) {
        restClient
                .get(url + "/status/" + id)
                .send(StatusResponse.class)
                .repeatWhen(completed -> completed.delayElements(Duration.ofMillis(2000)))
                .doOnNext(response -> {
                    if ("completed".equals(response.getStatus())) {
                        // 处理完成响应逻辑
                    }
                    // 也可以在这里添加中间状态的处理逻辑
                })
                .takeUntil(statusResponse->statusResponse.getStatus().equals("completed"))
                .doOnError(e -> log.error("Error :: " + e.getMessage(), e))
                .subscribe();
    }

• 方案三：在subscribe的onNext回调中处理
  也可以直接在订阅的元素消费回调中处理完成响应，无需额外添加算子：

private void status(String id) {
        restClient
                .get(url + "/status/" + id)
                .send(StatusResponse.class)
                .repeatWhen(completed -> completed.delayElements(Duration.ofMillis(2000)))
                .takeUntil(statusResponse->statusResponse.getStatus().equals("completed"))
                .doOnError(e -> log.error("Error :: " + e.getMessage(), e))
                .subscribe(response -> {
                    if ("completed".equals(response.getStatus())) {
                        // 编写完成响应的处理逻辑
                    }
                });
    }

内容的提问来源于stack exchange，提问作者Melad Basilius