所有示例代码：https://github.com/cumtbzy2011/webfluxdemo
功能与api
背景
Netty作为java领域首屈一指的nio框架，其以优越的性能被众多中间件所使用。但到了java的web开发领域，却很难享受到Netty的性能优势。其原因在于传统web开发基于servlet容器，许多依赖和开发框架都是基于servlet实现的，比如spring。而netty为了保持代码的简单和高效，并没有实现servlet标准，这就导致将web容器迁移到netty后许多框架和第三方库不能使用，迁移的成本过大。但spring webflux出现改变了这一现状。她在兼容原有mvc开发方式的同时，重写和实现了大量第三方库，在提升性能的同时，降低了迁移的成本。同时spring webflux适配多种web容器，即使仍然使用tomcat也是可以的。
接口声明
接口声明除了保留原有注解式声明的方式，为了满足reactor的编程风格，额外支持了函数式声明的方式。通工具类RouterFunctions过构造RounterFunction对象，并向Spring注入实现函数式接口声明。
@Bean
public TestHandler testHandler() {
return new TestHandler();
}
@Bean
public RouterFunction<ServerResponse> routes(TestHandler testHandler) {
return RouterFunctions.route(RequestPredicates.POST("/route"),
testHandler::echoName);
}
@GetMapping("anno")
public String sayHello(String name) {
return "hello world! " + name;
}
class TestHandler {
public Mono<ServerResponse> echoName(ServerRequest request) {
return request.bodyToMono(Post.class)
.map(Post::getName)
.flatMap(name -> ServerResponse.ok()
.contentType(MediaType.TEXT_PLAIN)
.body(BodyInserters.fromObject("hello world!" + name)));
}
}
在WebFlux中，request和respose不再是原来的ServletRequest和ServletRequest，取而代之的是ServerRequest和ServerResponse。这两个对象是webflux新出现的。首先webflux底层如果使用了reactor-netty，那么自然就没有所谓的servlet一说，另外ServerRequest和ServerResponse提供了对non-blocking和backpressure特性的支持，提供了将Http消息内容转换成Mono和Flux的方法，使响应式编程成为了可能。
过滤器Filter
过滤器的使用方法和spring mvc类似，不过与ServerRequest和ServerResponse相同的是，webflux提供了一个新的过滤器接口WebFilter以提供对Mono和Flux的支持。代码如下：
@Component
public class DemoWebFilter implements WebFilter{
@Override
public Mono<Void> filter(ServerWebExchange serverWebExchange, WebFilterChain webFilterChain) {
if (!serverWebExchange.getRequest().getHeaders().containsKey("token")) {
serverWebExchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
return Mono.empty();
}
return webFilterChain.filter(serverWebExchange);
}
}
值得注意的是Mono<Void>这个返回值，在框架的很多地方都会用到。他意味着一个空的Mono，对于任何读取，他会立刻发出一个complete信号。相比与直接返回void，Mono<Void>作为方法的返回值时，可以对该方法进行链式调用。另外虽然Mono<Void>虽然没有返回值，但是其本身的complete或者error状态，也可以注册回调进行异步处理。
异常处理
在Spring Webflux中，异常分两种。一是controller中方法抛出的异常，这在webflux中同样可以像在mvc中用@ExceptionHandler声明异常处理方法。二是在WebHandler API这种比较偏底层的api，典型的是WebFilter，异常处理使用了支持Mono的新接口：WebExceptionHandler，可用于处理来自WebFilter链和WebHandler的异常。使用WebExceptionHandler时，只要将其声明为Spring bean即可自动注入并使用，并可选择通过bean声明上的@Order或通过实现Ordered来表示优先级。需要注意的是webflux有默认的WebExceptionHandler-DefaultErrorWebExceptionHandler，其order为默认的-1。如果我们想自定义WebExceptionHandler，那么必须将order声明为-2以上，否则异常将不会传递到我们自定义的WebExceptionHandler中。
@Component
//要比DefaultErrorWebExceptionHandler优先级-1高
//比较底层，如果异常被@ExceptionHandler处理了，那么将不会由此处理
//可以处理filter和webHandler中的异常
@Order(-2)
public class ErrorLogHandler implements WebExceptionHandler {
@Override
public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
exchange.getResponse().setStatusCode(HttpStatus.OK);
byte[] bytes = ("ErrorLogHandler: " + ex.getMessage()).getBytes(StandardCharsets.UTF_8);
DataBuffer wrap = exchange.getResponse().bufferFactory().wrap(bytes);
return exchange.getResponse().writeWith(Flux.just(wrap));
}
}
@ExceptionHandler(Exception.class)
public String test(Exception e) {
return "@ExceptionHandler: " + e.getMessage();
}
Multipart和Stream
在基础框架reactor中Mono代表一个单次发送的数据源，而Flux代表一个可多次发送的数据源。在spring webflux的controller中，Mono很好理解，代表前端的一次传参或接口的一次返回。那么Flux该如何使用呢？简单来说Flux在这两个场景下使用：接受Multipart参数、返回Stream类型数据或者用于分批返回。代码如下：
@PostMapping(value = "", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
Mono<String> requestBodyFlux(@RequestBody Flux<Part> parts) {
return parts.map(part -> part instanceof FilePart
? part.name() + ":" + ((FilePart) part).filename()
: part.name())
.collect(Collectors.joining(",", "[", "]"));
}
//如果不是application/stream json則呼叫端無法滾動得到結果，將一直阻塞等待資料流結束或超時。
@GetMapping(value = "stream", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<Post> getBeanStream() {
return Flux.interval(Duration.ofMillis(500))
.map(l -> new Post("bian", LocalDateTime.now()))
.log();
}
Multipart是Htp请求的一种常见的数据结构，常用于表单提交。在spring mvc中，表单中的每个键值对会映射成一个个part。到了webflux，自然而然地转换成代表多个表单字段Flux。而返回值Flux，则对应了一种新的MediaType：APPLICATION_STREAM_JSON_VALUE。他的使用需要浏览器或者客户端的支持。从使用中来看，浏览器会对每一次返回的数据分批处理。如果简单的get调用，会在页面滚动打印返回值，直到Flux发射完成：
image.png
而如果接口并没有声明produces = MediaType.APPLICATION_STREAM_JSON_VALUE的媒体类型，浏览器将会在Flux所有数据发射完毕后一次性打印。
WebSocket
在webflux中使用WebSocket功能很简单，只要注册WebSocketHandlerAdapter用于websocket协议的握手，再定义对应路径的websocket消息处理器即可：
@Configuration
@ComponentScan
@EnableWebFlux
class WebConfig {
@Bean
public HandlerMapping handlerMapping() {
Map<String, WebSocketHandler> map = new HashMap<>();
map.put("/echo", new EchoWebSocketHandler());
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setUrlMap(map);
return mapping;
}
@Bean
WebSocketHandlerAdapter webSocketHandlerAdapter(){
return new WebSocketHandlerAdapter();
}
}
public class EchoWebSocketHandler implements WebSocketHandler {
public EchoWebSocketHandler() {
}
@Override
public Mono<Void> handle(WebSocketSession session) {
return session.send( //1. 向一个websocket连接发送一段消息
session.receive() //2. 获得入站消息流
.doOnNext( //3. 对每一个websocket消息进行处理，相当于stream的map，返回的仍是一个流
WebSocketMessage::retain //4. 保留消息（主要针对池化内存（内部使用了netty的ByteBuf），使之引用计数+1，避免过早回收）
)
);
}
}
需要注意的是，通过webSocketSession.receive() 获得的Flux<WebSocketMessage>，其每一次发射的数据WebSocketMessage如果是再Netty容器中，是一个对Netty中ByteBuf的保证，而ByteBuf在使用中有一点要注意，就是谁使用谁释放、retain()和release()成对出现。所以当把Flux<WebSocketMessage>发射的WebSocketMessage传递给其他方法使用时，注意要retain()增加一次计数，避免上一级方法release()使ByteBuf引用计数归零，导致过早回收。关于Netty的内存使用，下面会写一篇简要的介绍文章。
Mongo
MongoDB由于支持异步客户端，所以很适合在webflux项目中使用，spring-data-reactor也在第一时间做了支持。配合springboot的@EnableMongoAuditing注解，可以很快搭建异步mongo客户端。相关代码如下：
@SpringBootApplication
@EnableMongoAuditing
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
@Component
@Slf4j
class DataInitializer implements CommandLineRunner {
private final PostRepository posts;
public DataInitializer(PostRepository posts) {
this.posts = posts;
}
@Override
public void run(String[] args) {
log.info("start data initialization ...");
this.posts
.deleteAll()
.thenMany(
Flux
.just("bianzhaoyu", "xinan")
.flatMap(
name -> this.posts.save(Post.builder().name(name).age(25).build())
)
)
.log()
.subscribe(
null,
null,
() -> log.info("done initialization...")
);
}
}
@RestController()
@RequestMapping(value = "/posts")
class PostController {
private final PostRepository posts;
public PostController(PostRepository posts) {
this.posts = posts;
}
@GetMapping("")
public Flux<Post> all() {
return this.posts.findAll();
}
@PostMapping("")
public Mono<Post> create(@RequestBody Post post) {
return this.posts.save(post);
}
@GetMapping("/{id}")
public Mono<Post> get(@PathVariable("id") String id) {
return this.posts.findById(id);
}
@PutMapping("/{id}")
public Mono<Post> update(@PathVariable("id") String id, @RequestBody Post post) {
return this.posts.findById(id)
.map(p -> {
p.setName(post.getName());
p.setAge(post.getAge());
return p;
})
.flatMap(p -> this.posts.save(p));
}
@DeleteMapping("/{id}")
public Mono<Void> delete(@PathVariable("id") String id) {
return this.posts.deleteById(id);
}
}
interface PostRepository extends ReactiveMongoRepository<Post, String> {
}
@Data
@ToString
@Builder
@NoArgsConstructor
@AllArgsConstructor
class Post {
@Id
private String id;
private String name;
private Integer age;
@CreatedDate
private LocalDateTime createdDate;
}
配置如下：
spring:
data:
mongodb:
uri: mongodb://localhost:27017/blog
grid-fs-database: images
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
</dependencies>
Redis
异步Redis客户端的使用和普通Redis客户端类似，只是RedisTemplate的方法都原生支持了异步调用。使用时只要引入spring-boot-starter-data-redis-reactive依赖，并注册ReactiveRedisTemplate即可：
@Bean
public ReactiveRedisTemplate<String, Post> reactiveJsonPostRedisTemplate(
ReactiveRedisConnectionFactory connectionFactory) {
RedisSerializationContext<String, Post> serializationContext = RedisSerializationContext
.<String, Post>newSerializationContext(new StringRedisSerializer())
.hashKey(new StringRedisSerializer())
.hashValue(new Jackson2JsonRedisSerializer<>(Post.class))
.build();
return new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
}
@Component
class PostRepository {
ReactiveRedisOperations<String, Post> template;
public PostRepository(ReactiveRedisOperations<String, Post> template) {
this.template = template;
}
Flux<Post> findAll() {
return template.<String, Post>opsForHash().values("posts");
}
Mono<Post> findById(String id) {
return template.<String, Post>opsForHash().get("posts", id);
}
Mono<Post> save(Post post) {
if (post.getId() != null) {
String id = UUID.randomUUID().toString();
post.setId(id);
}
return template.<String, Post>opsForHash().put("posts", post.getId(), post)
.log()
.map(p -> post);
}
Mono<Void> deleteById(String id) {
return template.<String, Post>opsForHash().remove("posts", id)
.flatMap(p -> Mono.<Void>empty());
}
Mono<Boolean> deleteAll() {
return template.<String, Post>opsForHash().delete("posts");
}
}
MySQL
mysql作为现在使用最广的数据存储工具，可以说是选择任何框架时必须考虑到兼容性的一点。但是遗憾的是，由于JDBC协议只支持同步访问，spring目前并没有直接对jdbc的reactor客户端的支持。虽然可以通过引入第三方异步数据库连接池，或者将普通jpa方法用Mono，Flux指定调用线程池的方式进行包装，但是作为关系型数据库最重要的一点：事务，却无法用@Transactional实现。虽然可以将一个事务的代码写在一个异步函数中，但却无法做到像同步方法那样，使用@Transactional各个业务方法，导致可复用性和实用性极低。这里使用一个异步jdbc线程池rxjava2-jdbc，相比与Mono/Flux包装的方式，rxjava2-jdbc在返回一个connection时是异步的，虽然由于jdbc协议的线程，执行sql语句的时候仍然是同步阻塞的。rxjava-jdbc内部维护了一个线程池用于执行阻塞代码，这也避免了我们自定义线程池的麻烦。
pom依赖：
<dependency>
<groupId>com.github.davidmoten</groupId>
<artifactId>rxjava2-jdbc</artifactId>
<version>0.1-RC23</version>
</dependency>
代码如下：
/**
* spring-data-jpa是同步的，repository返回的结果并不是Mono或者Flux形式。
* 可以使用第三方异步jdbc连接池rxjava2-jdbc，但是由于每个方法是异步的，
* 当数个异步方法组合起来时，并不能保证每个方法都是由一个线程按顺序调用的，
* 这就使基于ThreadLocal的@Transactional无法使用
* 当然，可以手动在一个异步方法中开启并提交事务，但是这还是失去了@Transactional组合
* 不同方法到一个事物的便利性和可扩展性
* @author xinan
*/
@Component
public class RxJava2PostRepository {
private Database db;
RxJava2PostRepository(Database db) {
this.db = db;
}
public Observable<Post> findAll() {
return this.db.select("select * from posts")
.get(
rs -> new Post(rs.getLong("id"),
rs.getString("name"),
rs.getInt("age")
)
)
.toObservable();
}
public Single<Post> findById(Long id) {
return this.db.select("select * from posts where id=?")
.parameter(id)
.get(
rs -> new Post(rs.getLong("id"),
rs.getString("name"),
rs.getInt("age")
)
)
.firstElement()
.toSingle();
}
public Single<Integer> save(Post post) {
return this.db.update("insert into posts(name, age) values(?, ?)")
.parameters(post.getName(), post.getAge())
.returnGeneratedKeys()
.getAs(Integer.class)
.firstElement()
.toSingle();
}
String sql = "insert into posts(title, content) values(?, ?)";
//使用事务
public Single<Integer> saveTx(Post post) {
return db.connection()
.map(connection -> {
connection.setAutoCommit(false);
PreparedStatement pstmt = connection.prepareStatement(sql);
pstmt.setInt(1, post.getAge());
pstmt.setInt(2, post.getAge());
int i = pstmt.executeUpdate();
pstmt.close();
connection.commit();
return i;
});
}
}
网友评论