背景
基本概念
响应式编程关键词:
数据流:流式处理
异步: 异步处理
消息:基于消息名
Reactor模型
1. 客户端先向服务器端注册感兴趣的event,完成了事件订阅;
2. 客户端发生已经注册的事件,会触发服务器的响应,服务器存在一个selector线程,【轮询客户端发送过来的事件】但是并不实际处理事件,而是找到对应的Request Handler,启用另外一条线程运行处理。
3. 最终结果会转换成data stream,发送到客户端;
WebFlux
基于servlet3.1对非阻塞机制,和java8的函数式语法,webflux出现了。
响应式编程分为3层:
容器要求:支持servlet3.1
java异步编程领域:Netty
DispatcherHandler
跟springMVC对标
核心处理代码
public Mono<Void> handle(ServerWebExchange exchange) {
if (this.handlerMappings == null) {
return createNotFoundError();
}
return Flux.fromIterable(this.handlerMappings)
.concatMap(mapping -> mapping.getHandler(exchange))
.next()
.switchIfEmpty(createNotFoundError())
.flatMap(handler -> invokeHandler(exchange, handler))
.flatMap(result -> handleResult(exchange, result));
}
核心处理流程
注意
spring webflux只支持spring data reactive的数据源;
而数据库的开发往往是堵塞的,所以,spring data reactive并不能对数据库的开发提供支持。
适用于 redis,mongodb等nosql数据库
WebFlux的服务端开发
启动器代码
package com.springbootpractice.demo.webflux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
@EnableReactiveMongoRepositories(basePackages = "com.springbootpractice.demo.webflux.dao.repository")
public class DemoWebfluxApplication {
public static void main(String[] args) {
SpringApplication.run(DemoWebfluxApplication.class, args);
}
}
持久层代码
package com.springbootpractice.demo.webflux.dao.repository;
import com.springbootpractice.demo.webflux.dao.entity.User;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
/**
* 说明:TODO
* @author carter
* 创建时间: 2020年01月15日 6:18 下午
**/
@Repository
public interface UserRepository extends ReactiveMongoRepository<User,Long> {
Flux<User> findByUserNameLikeAndNoteLike(String userName,String note);
}
控制器代码
**
package com.springbootpractice.demo.webflux.controller;
import com.springbootpractice.demo.webflux.core.UserValidator;
import com.springbootpractice.demo.webflux.dao.entity.User;
import com.springbootpractice.demo.webflux.service.UserService;
import org.springframework.validation.DataBinder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.InitBinder;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import javax.validation.Valid;
/**
* 说明:TODO
* @author carter
* 创建时间: 2020年01月15日 6:43 下午
**/
@RestController
public class UserController {
private final UserService userService;
public UserController(UserService userService) {
this.userService = userService;
}
@GetMapping(path = "/user/{id}")
public Mono<User> getUser(@PathVariable("id") Long id){
return userService.getUserById(id);
}
@GetMapping(path = "/user/{userName}/{note}")
public Flux<User> getUser(@PathVariable("userName") String userName, @PathVariable("note") String note){
return userService.findByUserNameAndNote(userName,note);
}
@GetMapping(path = "/user/insert/{user}")
public Mono<User> insertUser(@Valid @PathVariable("user") User user){
return userService.insertUser(user);
}
// @InitBinder
// public void initBinder(DataBinder binder){
// binder.setValidator(new UserValidator());
// }
}
WebFlux的核心功能
类比springMVC,提供了 WebFluxConfigurer进行配置,根据需要实现对应的方法;
转换器
对标 springMVC,也需要实现Converter接口:
代码
package com.springbootpractice.demo.webflux.core;
import com.springbootpractice.demo.webflux.dao.entity.User;
import com.springbootpractice.demo.webflux.dao.entity.enums.SexEnum;
import org.springframework.core.convert.converter.Converter;
import org.springframework.util.Assert;
import java.util.Objects;
/**
* 说明:TODO
* @author carter
* 创建时间: 2020年01月16日 9:28 上午
**/
public class String2UserConverter implements Converter<String, User> {
@Override
public User convert(String s) {
final String[] split = Objects.requireNonNull(s,"转换为User的string不能为空").split("-");
Assert.isTrue(split.length==4,"转换为User的string必须含有4个字段");
return User.builder()
.id(Long.parseLong(split[0]))
.userName(split[1])
.note(split[2])
.sex(SexEnum.getSexEnum(Integer.parseInt(split[3])))
.build();
}
}
校验器
对标springMVC的校验器, 实现Validator接口;
代码
package com.springbootpractice.demo.webflux.core;
import com.springbootpractice.demo.webflux.dao.entity.User;
import org.apache.logging.log4j.util.Strings;
import org.springframework.validation.Errors;
import org.springframework.validation.Validator;
/**
* 说明:TODO
* @author carter
* 创建时间: 2020年01月16日 9:48 上午
**/
public class UserValidator implements Validator {
@Override
public boolean supports(Class<?> clazz) {
return clazz.equals(User.class);
}
@Override
public void validate(Object target, Errors errors) {
User user = (User) target;
if (Strings.isBlank(user.getUserName())){
errors.rejectValue("userName","","用户名不能为空");
}
}
}
局部校验器
对标springMVC, 可以在控制器中增加 # ,里面配置好本控制器的校验器
代码
@InitBinder
public void initBinder(DataBinder binder){
binder.setValidator(new UserValidator());
}
设置静态资源
一些文件,图片,配置内容的配置,可以在WebConfigurer中进行配置;
代码
package com.springbootpractice.demo.webflux.core;
import org.springframework.context.annotation.Configuration;
import org.springframework.format.FormatterRegistry;
import org.springframework.http.CacheControl;
import org.springframework.validation.Validator;
import org.springframework.web.reactive.config.ResourceHandlerRegistry;
import org.springframework.web.reactive.config.WebFluxConfigurer;
import java.util.concurrent.TimeUnit;
/**
* 说明:TODO
* @author carter
* 创建时间: 2020年01月16日 9:33 上午
**/
@Configuration
public class WebFluxConfig implements WebFluxConfigurer {
@Override
public void addFormatters(FormatterRegistry registry) {
registry.addConverter(new String2UserConverter());
}
@Override
public Validator getValidator() {
return new UserValidator();
}
@Override
public void addResourceHandlers(ResourceHandlerRegistry registry) {
registry.addResourceHandler("/static/**")
.addResourceLocations("classpath:/static/")
.setCacheControl(CacheControl.maxAge(1, TimeUnit.HOURS));
}
}
访问静态资源例子:
小结
通过本篇文章,你学会了:
1. springWebFlux的基础概念和reactor模型;
1. 一个利用webFlux操作mongodb的增删改查的例子;
1. 类比springMVC,开发webflux的转换器,校验器,静态资源;
https://github.com/carterbrother/springbootpractice/tree/master/demo_webflux
原创不易,转载请注明出处。