Reactor系列(十五)backpressure背压

package com.example.reactor;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.List;

@Slf4j
public class FluxMonoTestCase extends BaseTestCase {
    @Test
    public void pressure() throws InterruptedException {
        Flux<Long> longFlux = Flux.interval(Duration.ofMillis(1));
        longFlux.subscribe(new Subscriber<Long>() {
            Subscription subscription;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Long aLong) {
                perform(3L);
                subscription.request(3);
                System.out.println("val:"+aLong);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });
        Thread.sleep(1000);
    }
}