Spring响应式编程之Reactor背压机制

背压机制

在生产-发布模式下，由于事件处理平台大小有限制，生产者和消费者存在处理速度的差异，生产者不能持续不断的发布数据并推送，即流量控制问题。异步式编程采用背压机制来保持系统稳定。

背压机制简单理解，通过一种反馈机制来告知生产者当前消费者的处理能力，生产者动态调整数据生成的速率，确保消费者能够处理生产者产生的数据。

背压的策略包括：

缓冲（buffering）：将超出消费者处理能力的数据暂存在缓冲区中，等消费者有能力处理时再发送。这会实现无限缓冲，并可能导致OutOfMemoryError，因此一般适用于消费者偶尔会被短暂压垮，但整体处理能力足够栋场景；

丢弃（dropping）：丢弃超出消费者处理能力的数据，只保留最新的数据或最重要的数据。一般适用于消费者不关心部分数据的丢失；

最新（lastest）：只保留最新的数据，丢弃旧的数据，确保消费者始终处理最新的数据。一般适用于实时的场景；

错误（error）：当消费者无法处理数据时，抛出异常或错误；

请求（Request-based）：消费者根据自身处理能力向生产者请求数据，生产者只在接收到请求时才发送数据。适用于消费者处理能力波动较大，需要动态调整处理速率的场景。

缓冲示例：

Flux<Integer> source = Flux.range(1, 100);
source.onBackpressureBuffer(10) // 设置缓冲区大小为10.subscribe(System.out::println);

丢弃示例：

Flux<Integer> source = Flux.range(1, 100);
source.onBackpressureDrop(item -> System.out.println("Dropped: " + item)).subscribe(System.out::println);

最新示例：

Flux<Integer> source = Flux.range(1, 100);
source.onBackpressureLatest().subscribe(System.out::println);

错误示例：

Flux<Integer> source = Flux.range(1, 100);
source.onBackpressureError().subscribe(System.out::println,error -> System.err.println("Error: " + error));

请求示例：

Flux<Integer> source = Flux.range(1, 100);
source.subscribe(new BaseSubscriber<Integer>() {@Overrideprotected void hookOnSubscribe(Subscription subscription) {request(1); // 初始请求1个数据}@Overrideprotected void hookOnNext(Integer value) {System.out.println(value);// 每处理完一个数据后再请求一个数据request(1);}
});