1. cache,onBackpressureBuffer。都是缓存。cache可以将hot流的数据缓存起来。onBackpressureBuffer也是缓存,但是当下游消费者的处理速度比上游生产者慢时,上游生产的数据会被暂时存储在缓冲区中,防止丢失。
2. Flux.range 默认带有无界缓冲区
Flux.interval 默认下游的consume速率不能低于produce的速率,否则直接报错。但是可以增加onBackpressureBuffer/cache让数据缓存 3. Flux.fromIterable 经过测试,应该也是属于无界缓冲区
4. zipWith自带32长度的缓存,因此可以做到适配两个flux。如果其中一个flux超过了32的缓冲,就会报错
Flux<Integer> existingFlux = Flux.interval(Duration.ofMillis(1)).map(t-> Integer.valueOf(t.intValue())); Flux<Long> s2 = Flux.interval(Duration.ofHours(1));
existingFlux.log() .zipWith(s2, (httpData, existingData) -> existingData // 合并两个数据 ).subscribe(); Thread.sleep(1000*10);
报错:
4. 注意默认带有无界缓冲区时,如果设置了onBackpressureBuffer(size)。则反而成为了有界缓冲区。有界会导致如果缓冲区不足以缓冲数据时,会报错