Spring webflux

Spring 5.0 Spring webflux 是一个全新的非堵塞的函数式 Reactive Web 框架，可以用来构建异步的、非堵塞的、事件驱动的服务。
springboot2.0 发布不久，最近研究了一下 springboot2.0 的新特性, 其中就发现了 webflux。

下面是 spring-flux 的一个 demo 话不多少上代码

使用 webflux 和 MVC 的区别就是在 artifacId 后面加上 flux

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.0.RELEASE</version>
</parent>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

@RestController
public class HelloController {@GetMapping("/hello")
    public String hello() {return "hello world";}
}

在 webflux 中有 Handler 和 Router 的概念，分别与 springmvc 中的 controllerr 和 equest mapper 相对应, 通俗的将就是 handler 就是真正处理请求的 bean，可以在 handler 中编写处理请求的逻辑，而 Router 就是如何让请求找到对应的 handler 中的方法处理，下面我们来实现一个简单的 handler 和 router。

@Component
public class HelloWorldHandler {public Mono<ServerResponse> helloWorld(ServerRequest request){return ServerResponse.ok()
                .contentType(MediaType.TEXT_PLAIN)
                .body(BodyInserters.fromObject("hello flux"));
    }
    
}

上面是一个简单的 handler 只相应了一个“hello flux”字符串！

@Configuration
public class RouterConfig {

    @Autowired
    private HelloWorldHandler helloWorldHandler;

    @Bean
    public RouterFunction<?> helloRouter() {return RouterFunctions.route(RequestPredicates.GET("/hello"), helloWorldHandler::helloWorld);
    }

}

上面是对应的 router 对应的是匹配一个 get 方式的 /hello 请求，然后调用 helloWorldHandler 中的 helloWorld 方法向浏览器输出一个文本类型的字符串

再来一个例子

@Component
public class UserHandler {

    @Autowired
    private ReactiveRedisConnection connection;

    public Mono<ServerResponse> getTime(ServerRequest request) {return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("Now is" + new SimpleDateFormat("HH:mm:ss").format(new Date())), String.class);
    }
    public Mono<ServerResponse> getDate(ServerRequest request) {return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("Today is" + new SimpleDateFormat("yyyy-MM-dd").format(new Date())), String.class);
    }

    public Mono<ServerResponse> sendTimePerSec(ServerRequest request) {return ServerResponse.ok().contentType(MediaType.TEXT_EVENT_STREAM)
                .body(Flux.interval(Duration.ofSeconds(1)).map(l -> new SimpleDateFormat("HH:mm:ss").format(new Date())), String.class);
    }


    public Mono<ServerResponse> register(ServerRequest request) {Mono<Map> body = request.bodyToMono(Map.class);
        return body.flatMap(map -> {String username = (String) map.get("username");
            String password = (String) map.get("password");
            String hashedPassword = BCrypt.hashpw(password, BCrypt.gensalt());
            return connection.stringCommands()
                    .set(ByteBuffer.wrap(username.getBytes()), ByteBuffer.wrap(hashedPassword.getBytes()));
        }).flatMap(aBoolean -> {Map<String, String> result = new HashMap<>();
            ServerResponse serverResponse = null;
            if (aBoolean){result.put("message", "successful");
                return ServerResponse.ok()
                        .contentType(MediaType.APPLICATION_JSON_UTF8)
                        .body(BodyInserters.fromObject(result));
            }else {result.put("message", "failed");
                return ServerResponse.status(HttpStatus.BAD_REQUEST)
                        .contentType(MediaType.APPLICATION_JSON_UTF8)
                        .body(BodyInserters.fromObject(request));
            }
        });

    }

    public Mono<ServerResponse> login(ServerRequest request) {Mono<Map> body = request.bodyToMono(Map.class);
        return body.flatMap(map -> {String username = (String) map.get("username");
            String password = (String) map.get("password");
            return connection.stringCommands().get(ByteBuffer.wrap(username.getBytes())).flatMap(byteBuffer -> {byte[] bytes = new byte[byteBuffer.remaining()];
                byteBuffer.get(bytes, 0, bytes.length);
                String hashedPassword = null;
                try {hashedPassword = new String(bytes, "UTF-8");
                } catch (UnsupportedEncodingException e) {e.printStackTrace();
                }
                Map<String, String> result = new HashMap<>();
                if (hashedPassword == null || !BCrypt.checkpw(password, hashedPassword)) {result.put("message", "账号或密码错误");
                    return ServerResponse.status(HttpStatus.UNAUTHORIZED)
                            .contentType(MediaType.APPLICATION_JSON_UTF8)
                            .body(BodyInserters.fromObject(result));
                } else {result.put("token", "无效 token");
                    return ServerResponse.ok()
                            .contentType(MediaType.APPLICATION_JSON_UTF8)
                            .body(BodyInserters.fromObject(result));
                }
            });
        });
    }


}

@Configuration
public class RouterConfig {

    @Autowired
    private HelloWorldHandler helloWorldHandler;

    @Bean
    public RouterFunction<?> helloRouter() {return RouterFunctions.route(RequestPredicates.GET("/hello"), helloWorldHandler::helloWorld);
    }

    @Autowired
    private UserHandler userHandler;

    @Bean
    public RouterFunction<ServerResponse> timerRouter() {return RouterFunctions.route(RequestPredicates.GET("/time"), userHandler::getTime)
                .andRoute(RequestPredicates.GET("/date"), userHandler::getDate);
    }

    @Bean
    public RouterFunction<?> routerFunction() {return RouterFunctions.route(RequestPredicates.GET("/hello"), helloWorldHandler::helloWorld)
                .andRoute(RequestPredicates.POST("/register"), userHandler::register)
                .andRoute(RequestPredicates.POST("/login"), userHandler::login)
                .andRoute(RequestPredicates.GET("/times"), userHandler::sendTimePerSec);
    }

}