public interface Publisher<T> {
    void subscribe(Subscriber<? super T> var1);
}

public interface Subscriber<T> {
? ? void onSubscribe(Subscription var1);?
? ? void onNext(T var1);?
? ? void onError(Throwable var1);?
? ? void onComplete();
}

public interface Subscription {
? ? void request(long var1);?
? ? void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

<dependency>
? ? <groupId>org.springframework.boot</groupId>
? ? <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest { 
    @Test
    public void testReactor(){
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
        Mono<Integer> mono = Mono.just(1);
 
        Integer[] arr = {1,2,3,4,5,6};
        Flux<Integer> flux1 = Flux.fromArray(arr);
 
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6);
        Flux<Integer> flux2 = Flux.fromIterable(list);
 
        Flux<Integer> flux3 = Flux.from(flux);
 
        Flux<Integer> flux4 = Flux.fromStream(Stream.of(1, 2, 3, 4, 5, 6));
 
        flux.subscribe();
 
        flux1.subscribe(System.out::println);
 
        flux2.subscribe(System.out::println,System.err::println);
 
        flux3.subscribe(System.out::println,System.err::println,() -> System.out.println("complete"));
 
        flux4.subscribe(System.out::println,System.err::println,
                () -> System.out.println("complete"),
                subscription -> subscription.request(3)); 
        flux4.subscribe(new DemoSubscriber());
    }
 
    class DemoSubscriber extends BaseSubscriber<Integer>{
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribe");
            subscription.request(1);
        }
 
        @Override
        protected void hookOnNext(Integer value) {
            if(value == 4){
                //背壓，通知數(shù)據(jù)源，不要發(fā)送數(shù)據(jù)了
                cancel();
            }
            System.out.println(value);
            request(1);
        }
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest { 
    @Test
    public void testReactor() throws InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
 
        flux.map(i -> {
            System.out.println(Thread.currentThread().getName()+"-map1");
            return i * 3;
        }).publishOn(Schedulers.elastic()).map(
                i -> {
                    System.out.println(Thread.currentThread().getName()+"-map2");
                    return i / 3;
                }
        ).subscribeOn(Schedulers.parallel())
                .subscribe(i -> System.out.println(Thread.currentThread().getName()+"-" + i)); 
        Thread.sleep(10000);
    }
}

@RestController
public class DemoController { 
    @GetMapping("/demo")
    public Mono<String> demo(){
        return Mono.just("demo");
    }
}

@Component
public class DemoHandler { 
    public Mono<ServerResponse> hello(ServerRequest request){
        return ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("hello"),String.class);
    }
 
    public Mono<ServerResponse> world(ServerRequest request){
        return ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("world"),String.class);
    }
 
    public Mono<ServerResponse> times(ServerRequest request){
        //每隔一秒發(fā)送當(dāng)前的時間
        return ok().contentType(MediaType.TEXT_EVENT_STREAM)
                .body(Flux.interval(Duration.ofSeconds(1))
                        .map(it -> new SimpleDateFormat("HH:mm:ss").format(new Date())),String.class);
    }
}

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函數(shù)的編寫
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times);
    }
}

@Component
public class DemoHandler { 
    @Autowired
    private PersonService personService; 
    public Mono<ServerResponse> queryPerson(ServerRequest request){
        Integer id = Integer.valueOf(request.pathVariable("id"));
        return ok().contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(Mono.just(personService.getPersonById(id)), Person.class);
    }
}

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函數(shù)的編寫
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times)
                .andRoute(GET("/queryPerson/{id}"),demoHandler::queryPerson);
    }
}

<dependency>
? ? <groupId>org.springframework.boot</groupId>
? ? <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>

#mongodb
spring.data.mongodb.uri=mongodb://root:yibo@localhost:27017
spring.data.mongodb.database=webflux

@Document(collection = "user")
@Data
public class User { 
    @Id
    private String id; 
    private String name; 
    private int age;
}
 
@Repository
public interface UserRepository extends ReactiveMongoRepository<User,String> {
}
 
@Component
public class DemoHandler { 
    @Autowired
    private UserRepository userRepository; 
    public Mono<ServerResponse> listUser(ServerRequest request){
        return ok().contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(userRepository.findAll(), User.class);
    }
 
    public Mono<ServerResponse> saveUser(ServerRequest request){
        String name = request.pathVariable("name");
        Integer age = Integer.valueOf(request.pathVariable("age"));
        User user = new User();
        user.setName(name);
        user.setAge(age);
        Mono<User> mono = Mono.just(user);
        return ok().build(userRepository.insert(mono).then());
    }
}

@Configuration
public class RouterConfig {?
? ? @Autowired
? ? private DemoHandler demoHandler;
?
? ? @Bean
? ? public RouterFunction<ServerResponse> demoRouter(){
? ? ? ? //路由函數(shù)的編寫
? ? ? ? return route(GET("/hello"),demoHandler::hello)
? ? ? ? ? ? ? ? .andRoute(GET("/world"),demoHandler::world)
? ? ? ? ? ? ? ? .andRoute(GET("/times"),demoHandler::times)
? ? ? ? ? ? ? ? .andRoute(GET("/queryPerson/{id}"),demoHandler::queryPerson)
? ? ? ? ? ? ? ? .andRoute(GET("/listUser"),demoHandler::listUser)
? ? ? ? ? ? ? ? .andRoute(GET("/saveUser/{name}/{age}"),demoHandler::saveUser);
? ? }
}

public class DispatcherHandler implements WebHandler, ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        initStrategies(applicationContext);
    }
}

protected void initStrategies(ApplicationContext context) {
    //獲取HandlerMapping及其子類型的bean
    //HandlerMapping根據(jù)請求request獲取handler執(zhí)行鏈
    Map<String, HandlerMapping> mappingBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerMapping.class, true, false);
 
    ArrayList<HandlerMapping> mappings = new ArrayList<>(mappingBeans.values());
    //排序
    AnnotationAwareOrderComparator.sort(mappings);
    this.handlerMappings = Collections.unmodifiableList(mappings);
 
    //獲取HandlerAdapter及其子類型的bean
    Map<String, HandlerAdapter> adapterBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerAdapter.class, true, false);
 
    this.handlerAdapters = new ArrayList<>(adapterBeans.values());
    //排序
    AnnotationAwareOrderComparator.sort(this.handlerAdapters);
 
    //獲取HandlerResultHandler及其子類型的bean
    Map<String, HandlerResultHandler> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerResultHandler.class, true, false);
 
    this.resultHandlers = new ArrayList<>(beans.values());
    AnnotationAwareOrderComparator.sort(this.resultHandlers);
}

public class RouterFunctionMapping extends AbstractHandlerMapping implements InitializingBean {
 
    @Nullable
    private RouterFunction<?> routerFunction;
 
    //讀取http傳輸數(shù)據(jù)，并解碼成一個對象
    private List<HttpMessageReader<?>> messageReaders = Collections.emptyList();
 
    public RouterFunctionMapping(RouterFunction<?> routerFunction) {
        this.routerFunction = routerFunction;
    }
 
    @Nullable
    public RouterFunction<?> getRouterFunction() {
        return this.routerFunction;
    }
 
    public void setMessageReaders(List<HttpMessageReader<?>> messageReaders) {
        this.messageReaders = messageReaders;
    }
 
    @Override
    public void afterPropertiesSet() throws Exception {
        if (CollectionUtils.isEmpty(this.messageReaders)) {
            ServerCodecConfigurer codecConfigurer = ServerCodecConfigurer.create();
            this.messageReaders = codecConfigurer.getReaders();
        }
 
        if (this.routerFunction == null) {
            //afterPropertiesSet方法調(diào)用的時候，routerFunction為null
            initRouterFunctions();
        }
    }
 
    protected void initRouterFunctions() {
        //獲取routerFunctions集合
        List<RouterFunction<?>> routerFunctions = routerFunctions();
         //將一個請求中含有多個路由請求RouterFunction合并成一個RouterFunction
        this.routerFunction = routerFunctions.stream().reduce(RouterFunction::andOther).orElse(null);
        logRouterFunctions(routerFunctions);
    }
 
    private List<RouterFunction<?>> routerFunctions() {
        //obtainApplicationContext()獲取ApplicationContext對象
        List<RouterFunction<?>> functions = obtainApplicationContext()
                //獲取指定bean的提供者，即上文配置的路由類
                .getBeanProvider(RouterFunction.class)
                //排序
                .orderedStream()
                //將流里面的都強(qiáng)轉(zhuǎn)成RouterFunction對象
                .map(router -> (RouterFunction<?>)router)
                .collect(Collectors.toList());
        return (!CollectionUtils.isEmpty(functions) ? functions : Collections.emptyList());
    }
 
    private void logRouterFunctions(List<RouterFunction<?>> routerFunctions) {
        //判斷當(dāng)前的日志級別是否是Debug
        if (logger.isDebugEnabled()) {
            int total = routerFunctions.size();
            String message = total + " RouterFunction(s) in " + formatMappingName();
            if (logger.isTraceEnabled()) {
                if (total > 0) {
                    routerFunctions.forEach(routerFunction -> logger.trace("Mapped " + routerFunction));
                }
                else {
                    logger.trace(message);
                }
            }
            else if (total > 0) {
                logger.debug(message);
            }
        }
    }
    ......
}

@Override
public Mono<Void> handle(ServerWebExchange exchange) {
    //handlerMappings在initStrategies()方法中已經(jīng)構(gòu)造好了
    if (this.handlerMappings == null) {
        return createNotFoundError();
    }
    //構(gòu)造Flux，數(shù)據(jù)源為handlerMappings集合
    return Flux.fromIterable(this.handlerMappings)
            //獲取Mono<Handler>對象，通過concatMap保證順序和handlerMappings順序一致
            //嚴(yán)格保證順序是因為在一個系統(tǒng)中可能存在一個Url有多個能夠處理的HandlerMapping的情況
            .concatMap(mapping -> mapping.getHandler(exchange))
            .next()
            //如果next()娶不到值則拋出錯誤
            .switchIfEmpty(createNotFoundError())
            //觸發(fā)HandlerApter的handle方法
            .flatMap(handler -> invokeHandler(exchange, handler))
            //觸發(fā)HandlerResultHandler 的handleResult方法
            .flatMap(result -> handleResult(exchange, result));
}

private Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
    return getResultHandler(result).handleResult(exchange, result)
            .onErrorResume(ex -> result.applyExceptionHandler(ex).flatMap(exceptionResult ->
                    getResultHandler(exceptionResult).handleResult(exchange, exceptionResult)));
}
 
private HandlerResultHandler getResultHandler(HandlerResult handlerResult) {
    if (this.resultHandlers != null) {
        for (HandlerResultHandler resultHandler : this.resultHandlers) {
            if (resultHandler.supports(handlerResult)) {
                return resultHandler;
            }
        }
    }
    throw new IllegalStateException("No HandlerResultHandler for " + handlerResult.getReturnValue());
}