如何实现每次仅发射1个事件、待订阅者处理完成再下发的Reactor Sink

你需要的「单订阅者串行处理+订阅默认值+速率控制」能力，原生Reactor没有直接的Processor实现，可以通过组合Replay Sink和串行消费逻辑实现，完全适配你提到的WebFlux中继服务场景。

核心改造思路 #

• 用Replay Sink存储全局的熔断状态，保证新请求能拿到最新的可用状态
• 所有请求先进入统一的队列排队，通过flatMap的并发参数强制串行消费，同一时间只发一个请求到下游服务
• 429触发时更新熔断状态，所有排队的请求等待直到状态恢复后再继续执行

完整实现代码 #

1. 定义全局公共组件 #

import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import java.time.Duration;

// 全局熔断状态Sink：保存最新的下游可用状态，默认初始为可用
Sinks.Many<Boolean> statusSink = Sinks.many().replay().latestOrDefault(true);
// 状态流做去重，只在状态变化时通知
Flux<Boolean> statusFlux = statusSink.asFlux().distinctUntilChanged();

// 全局请求队列Sink：所有来自端点A的请求先进入队列排队，支持背压缓冲
Sinks.Many<Mono<ResponseEntity<String>>> requestQueue = Sinks.many().unicast().onBackpressureBuffer();

// 启动全局串行消费进程，服务生命周期内只启动一次
requestQueue.asFlux()
        // 每次处理前先等待可用状态为true
        .flatMap(requestMono -> statusFlux.filter(available -> available).next().thenReturn(requestMono), 1)
        // 串行执行转发请求，flatMap第二个参数1强制同一时间只处理1个请求
        .flatMap(requestMono -> requestMono, 1)
        // 捕获429异常触发熔断逻辑
        .doOnError(WebClientResponseException.TooManyRequests.class, e -> {
            // 下发熔断状态
            statusSink.tryEmitNext(false);
            // 读取响应头的重试等待时长，单位秒
            int retryAfter = Integer.parseInt(e.getHeaders().getFirst("Retry-After"));
            // 等待结束后恢复可用状态
            Mono.delay(Duration.ofSeconds(retryAfter))
                    .doOnNext(unused -> statusSink.tryEmitNext(true))
                    .subscribe();
        })
        // 异常不中断消费链，处理其他业务异常
        .onErrorContinue((throwable, request) -> {
            // 这里可以加你的异常日志、告警逻辑
        })
        .subscribe();

2. 端点A的接口实现 #

@GetMapping("/proxy")
public Mono<ResponseEntity<String>> proxyRequest() {
    // 构造冷流的转发请求，订阅后才会实际执行
    Mono<ResponseEntity<String>> forwardRequest = webClientBuilder.build()
            .get()
            .uri("下游微服务B的请求地址")
            .retrieve()
            .toEntity(String.class);

    // 请求入队并返回响应结果
    return Mono.create(sink -> {
        // 把请求提交到队列
        requestQueue.tryEmitNext(forwardRequest)
                .orElseThrow(() -> new RuntimeException("服务繁忙，请稍后重试"));
        // 订阅请求结果，返回给调用方
        forwardRequest.subscribe(sink::success, sink::error);
    });
}

原有代码问题说明 #

你之前的实现会批量触发请求的原因：

• Sinks.many().replay()是多播模式，所有订阅者订阅后会立刻拿到最新值，5个线程同时订阅自然会同时触发请求
• 没有限制消费并发数，默认flatMap并发数为256，所有请求会同时执行

提示：如果你的服务需要支持分布式部署，可以把内存队列替换为RocketMQ、Kafka等分布式消息队列，熔断状态可以用Redis存储来实现多实例共享。

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