Reactor系列(四)subscribe訂閱

#java# #reactor# #subcribe# #訂閱#

視頻講解 :https://www.bilibili.com/video/av79117693/

FluxMonoTestCase.java

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;

@Slf4j
public class FluxMonoTestCase extends BaseTestCase {
    @Test
    public void subscribe(){
        Flux<String> stringFlux = Flux.just("Hello","World");
        //stringFlux.subscribe(System.out::println);
        //訂閱方式一
        stringFlux.subscribe(val ->{
            log.info("val:{}",val);
        },error ->{
            log.error("error:{}",error);
        },() ->{
            log.info("Finished");
        },subscription -> {
            subscription.request(1);
        });
        //訂閱方式二
        stringFlux.subscribe(new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(Long.MAX_VALUE);
            }
            @Override
            public void onNext(String s) {
                log.info("onNext:{}",s);
            }
            @Override
            public void onError(Throwable throwable) {
            }
            @Override
            public void onComplete() {
                log.info("onComplete");
            }
        });
    }
}

BaseTestCase.java

package com.example.reactor;

import java.util.Arrays;
import java.util.List;

public class BaseTestCase {
    protected static final List<Employee> list = Arrays.asList(
            new Employee(1, "Alex", 1000),
            new Employee(2, "Michael", 2000),
            new Employee(3, "Jack", 1500),
            new Employee(4, "Owen", 1500),
            new Employee(5, "Denny", 2000));
}

關注公眾號，堅持每天3分鍾視頻學習