本文主要是介绍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);}
});
这篇关于Spring响应式编程之Reactor背压机制的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!