1.响应式编程

  WebFlux是从Spring Framework 5.0开始引入响应式Web框架的。与Spring MVC不同, WebFlux不需要Servlet API,在完全异步且无阻塞,并通过Reactor项目实现Reactive Streams 规范。

  WebFlux可以在资源有限的情况下提高系统的吞吐量和伸缩性(不是提高性能)。这意味着, 在资源相同的情况下,WebFlux可以处理更多的请求(不是业务)。

  WebFlux除支持RESTful Web服务外,还可以用于提供动态HTML内容。

   MVC:

  MVC的工作流程是:主线程接收到请求(request)-准备数据-返回数据。整个过程是单线程咀塞的,用户会感觉等待时间长是因为,在结果处理好之后才返回数据给浏 览器。因此,如果请求很多,则吞吐量就上不去。

  WebFlux:

  WebFlux的工作流程是:主线程接收到请求—立刻返回数据与函数的组合(Mon或Flux, 不是结果),开启一个新Work线程去做实际的数据准备工作,进行真正的业务操作,Work 线程完成工作,返给用户真实数据(结果)。

  这种方式给人的感売是响应时间很短,因为返回的是不变的常数,它不随用户数量的增加而变化。

两者对比:

2.Mono和Flux

Mono和Flux是Reactor中的两个基本概念。

  • Mono和Flux属于事件发布者,为消费者提供订阅接口。当有事件发生时,Mono或Flux 会回调消费者的相应方法,然后通知消费者相应的事件。这也是晌应式编程模型。
  • Mono和Flux用于处理异步数据流,它不像MVC中那样直接返回String/List,而是将异步数据流包装成Mono或Flux对象

Mono和Flux的区别

  1. Flux可以发送很多item,并且这些item可以经过若干算子(operators )后才被订阅。 Mono只能发送一个itemo

  2. Mono主要用于返回单个数据。Flux用于返回多个数据。如果要根据id查询某个User对象,则返回的肯定是单个User,那么需要将其包装成Mono,若需要获取所有User (这是一个集合),则需要将这个集合包装成Flux,这里的单个数据并不是指一个数据,而是指封装好的一个对象。多个数据就是多个对象。

  3. Mono表示包含0或1个元素的异步序列。在该序列中可以包含3种不同类型的消息通知: 正常的包含元素的消息、序列结束的消息、序列岀错的消息。当消息通知(正常的包含元素的消息、 序列结束的消息、序列出错的消息)产生时,订阅者中有对应的方法onNext( )、onComplete( ). onError( )被调用

  4. Flux表示的是包含0到N个元素的异步序列,在该序列中可以包含与Mono相同的3种类型的消息通知。

  5. Flux和Mono之间可以进行转换。对一个Flux序列进行计数操作时,得到的结果是一个 Mono对象。把多个Mon。序列合并在一起,得到的是一个Flux对象

开发WebFlux的流程

  1. 注解式开发流程

  WebFlux是响应式框架,其中使用的注解式开发方式只是Spring团队为了更好地迁移而提供的。和MVC开发模式一样,地址映射也是通过@RequestMapping提供的,用@Controller或 @RestController 来代替 Handler 类。

  2.响应式开发流程  

    (1)创建 Handler 类

    这里的Handler类相当于Spring MVC的Controller层中的方法体。在响应式编程中,请求和响应不再是 httpsServletRequest 和 httpsServletResponse,而是变成了 ServerRequest 和 ServerResponse

    (2)配置 RouterFunction

    RouterFunction 和注解@RequestMapping 相似,都用于提供 URL 路径。RouterFunction 的格式也是固定的,第1个参数代表路径,第2个参数代表方法,合起来代表将URL映射到方法。

实例:Hello world

配置WebFlux依赖:

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

    编写控制器:

@RestController
public class HelloWorldController {
@GetMapping("/helloworld")
public Mono<String> helloworld(){
return Mono.just("这是响应式编程");
}
}

实例:用注解式开发实现数据的增加、删除、修改和查询

创建实体类:

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User {
private long id;
private String name;
private int age;
}

编写控制器:

package com.itheima.controller;
import com.itheima.domain.User;
import org.springframework.https.httpsStatus;
import org.springframework.https.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/user")
public class UserController {
Map<Long, User> users = new HashMap<Long, User>();
@PostConstruct
public void init() throws Exception {
users.put(Long.valueOf(1),new User(1,"John",28));
users.put(Long.valueOf(2),new User(2,"ane",29));
}
@GetMapping("/list")
public Flux<User> getAll() throws Exception {
return Flux.fromIterable(users.entrySet().stream()
.map(entry -> entry.getValue())
.collect(Collectors.toList()));
}
@GetMapping("/{id}")
public Mono<User> getUserById(@PathVariable Long id) throws Exception {
return Mono.justOrEmpty(users.get(id));
}
@PostMapping("")
public Mono<ResponseEntity<String>> addUser(User user) throws Exception {
users.put(user.getId(), user);
return Mono.just(new ResponseEntity<>("添加成功", httpsStatus.CREATED));
}
@PutMapping("/{id}")
public Mono<ResponseEntity<User>> updateUser(@PathVariable Long id,User user) throws Exception {
user.setId(id);
users.put(id, user);
return Mono.just(new ResponseEntity<>(user,httpsStatus.CREATED));
}
@DeleteMapping("/{id}")
public Mono<ResponseEntity<String>> deleteUser(@PathVariable Long id){
users.remove(id);
return Mono.just(new ResponseEntity<>("删除成功",httpsStatus.ACCEPTED));
}
}

看下数据转换过程:

Map<Long, User> users:{1=User(id=1, name=John, age=28), 2=User(id=2, name=ane, age=29)}
users.entrySet():[1=User(id=1, name=John, age=28), 2=User(id=2, name=ane, age=29)]
users.entrySet().stream().toString():java.util.stream.ReferencePipeline$Head@785a5991
users.entrySet().stream()
.map(entry -> entry.getValue()).toString():java.util.stream.ReferencePipeline$3@748631e4
users.entrySet().stream()
.map(entry -> entry.getValue())
.collect(Collectors.toList()):[User(id=1, name=John, age=28), User(id=2, name=ane, age=29)]
可以看出entrySet()将map转为了set集合,entrySet() 方法返回映射中包含的映射的 Set 视图。

对应数据类型:

Map<Long, User> users:type:class java.util.HashMap
users.entrySet():type:class java.util.HashMap$EntrySet
users.entrySet().stream():type:class java.util.stream.ReferencePipeline$Head
users.entrySet().stream()
.map(entry -> entry.getValue()):type:class java.util.stream.ReferencePipeline$3
users.entrySet().stream()
.map(entry -> entry.getValue())
.collect(Collectors.toList()):type:class java.util.ArrayList

users.entrySet().stream()
返回支持顺序和并行聚合操作的元素序列。以下示例说明了使用Stream和IntStream的聚合操作:
int sum = widgets.stream()
.filter(w -> w.getColor() == RED)
.mapToInt(w -> w.getWeight())
.sum();

  在此示例中, widgets是Collection。我们通过Collection.stream()创建一个Widget对象流,对其进行过滤以生成仅包含红色小部件的流,然后将其转换为表示每个红色小部件重量的int值流。然后将该流相加以产生总重量。除了作为对象引用流的Stream之外,还有IntStream 、 LongStream和DoubleStream的原始特化,所有这些都被称为“流”,并符合此处描述的特征和限制。为了执行计算,流操作被组合成一个流管道。流管道由源(可能是数组、集合、生成器函数、I/O 通道等)、零个或多个中间操作(将流转换为另一个流,例如filter(Predicate) ) 和终端操作(产生结果或副作用,例如count()或forEach(Consumer) )。流是懒惰的;仅在发起终端操作时才对源数据进行计算,并且仅在需要时消耗源元素。
  集合和流虽然有一些表面上的相似之处,但有不同的目标。馆藏主要关注其元素的有效管理和访问。相比之下,流不提供直接访问或操作其元素的方法,而是关注以声明方式描述其源以及将在该源上聚合执行的计算操作。但是,如果提供的流操作不提供所需的功能,则可以使用iterator()和spliterator()操作来执行受控遍历。
流管道,如上面的“小部件”示例,可以被视为对流源的查询。除非源明确设计用于并发修改(例如ConcurrentHashMap ),否则在查询流源时修改流源可能会导致不可预测或错误的行为。
大多数流操作接受描述用户指定行为的参数,例如上面示例中传递给mapToInt的 lambda 表达式w -> w.getWeight() 。

  为了保持正确的行为,这些行为参数:必须是无干扰的(它们不会修改流源);和在大多数情况下必须是无状态的(它们的结果不应依赖于在流管道执行期间可能更改的任何状态)。此类参数始终是函数接口的实例,例如Function ,并且通常是 lambda 表达式或方法引用。除非另有说明,否则这些参数必须为非 null 。一个流应该只被操作一次(调用一个中间或终端流操作)。例如,这排除了“分叉”流,其中相同的源提供两个或多个管道,或同一流的多次遍历。如果流实现检测到流正在被重用,它可能会抛出IllegalStateException 。但是,由于某些流操作可能返回其接收者而不是新的流对象,因此可能无法在所有情况下检测重用。流有一个close()方法并实现了AutoCloseable ,但几乎所有的流实例在使用后实际上并不需要关闭。通常,只有源为 IO 通道的流(例如由Files.lines(Path, Charset)返回的流)才需要关闭。大多数流由集合、数组或生成函数支持,不需要特殊的资源管理。 (如果流确实需要关闭,可以在try -with-resources 语句中将其声明为资源。)流管道可以按顺序或并行执行。这种执行模式是流的属性。流是通过初始选择顺序或并行执行来创建的。 (例如, Collection.stream()创建一个顺序流, Collection.parallelStream()创建一个并行流。)这种执行模式的选择可以通过sequential()或parallel()方法进行修改,并且可以通过以下方式查询isParallel()方法。

    users.entrySet().stream()
.map(entry -> entry.getValue())
.collect(Collectors.toList()):

  例如上方代码将返回map中包含的映射的set视图(即set集合),然后创建一个set引用流,将引用流中的数据取出,然后使用Collector对此流的元素执行可变归约操作,最后效果为将数据取出,并去掉了键,保留了值,转为了List数组。

   Collector封装了用作collect(Supplier, BiConsumer, BiConsumer)参数的函数,允许重用收集策略和组合收集操作,例如多级分组或分区。
  如果流是并行的,并且Collector是concurrent的,并且流是无序的或收集器是unordered的,则将执行并发归约(有关并发归约的详细信息,请参阅Collector 。)
这是终端操作。
  当并行执行时,可以实例化、填充和合并多个中间结果,以保持可变数据结构的隔离。因此,即使与非线程安全的数据结构(例如ArrayList )并行执行,也不需要额外的同步来进行并行归约。

  最后返回collect

测试API功能

(1)获取数据

  启动项目后访问URL:https://localhost:8080/user/list ,会得到两个初始化的数据,输岀结果:[{“id”:1,”name”:”John”,”age”:28},{“id”:2,”name”:”ane”,”age”:29}]

  访问URL:https://localhost:8080/user/1,得到的是id为1的单个对象,见输出结果:{"id":1,"name":"John","age":28}

(2)修改数据

  修改数据的方法是通过PUT方式访问”https://localhost:8080/user/1",提交相应的name和age字段来修改内容(使用的是测试工具Postman ),如下所示。稍后会提示修改成功。

  

(3)添加数据

  可以通过POST方式提_交_User对象来添加数据。

  

(4)删除数据

  

实例:用晌应式开发方式开发WebFlux

(1)编写处理器类Handler

Handler相当于MVC中的Controller,用于提供实现功能的方法,代码如下:

package com.itheima.handler;
import org.springframework.https.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;

@Component
public class HelloWorldHandler {
public Mono<ServerResponse>sayHello(ServerRequest serverRequest){
return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN).body(Mono.just("你好"),String.class);
}
}

(2)编写路由器类Router

Router的主要功能是提供路由映射,相当于MVC模式中的注解@RequestMapping

package com.itheima.router;
import com.itheima.handler.HelloWorldHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;
@Configuration
public class Router {
@Autowired
private HelloWorldHandler helloWorldHandler;
@Bean
public RouterFunction<ServerResponse> getString(){
return route(GET("/helloWorld"),req->helloWorldHandler.sayHello(req));
}
}

上述代码中,通过return route(GET("/helloWorld"),req->helloWorldHandler.sayHello(req)); 来指定路由,包含https方法和对应的功能方法。

实例:用WebFlux模式操作MongoDB数据库,实现数据的增加、删除、修改和查询的功能

(1)添加依赖

    要操作数据库,则需要添加相应的依赖。可以通过Spring Boot集成的MongoDB的Starter 依赖来快速实现配置和操作。具体依赖见以下代码

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>

在配置文件中配置MongoDB的地址信息(MongoDB 2.4以上版本),见以下代码:

spring.data.mongodb.uri=mongodb://127.0.0.1:27017/test

配置Mongo的格式如下:

spring.data.mongodb.uri=mongodb://用户名: 密码@ip 地址:端口号/数据库

(2)创建实体类

    这里编写实体类并没有特别需要讲解的,只是利用了 Lombok插代码,具体见以下代码:

package com.itheima.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User {
private long id;
private String name;
private int age;
}

(3)编写接口

    Spring Boot 的 Starter 提供了 ReactiveMongoRepository 接口,用于操作 Mongo 数据库, 用法见以下代码:  

package com.itheima.dao;
import com.itheima.domain.User;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;

public interface UserRepository extends ReactiveMongoRepository<User,String> {
}

(4)编写API

package com.itheima.controller;
import com.itheima.dao.UserRepository;
import com.itheima.domain.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import org.springframework.https.httpsStatus;
import org.springframework.https.MediaType;
import org.springframework.https.ResponseEntity;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import javax.annotation.PostConstruct;
import javax.validation.Valid;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/user")
public class UserController {
@Autowired
private UserRepository userRepository;
@GetMapping("/list")
public Flux<User> getAll(){
return userRepository.findAll();
}
@GetMapping(value = "/listdelay",produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<User> getAllDelay(){
return userRepository.findAll().delayElements(Duration.ofSeconds(1));
}
@GetMapping("/{id}")
public Mono<ResponseEntity<User>> getUser(@PathVariable String id){
return userRepository.findById(id)
.map(getUser->ResponseEntity.ok(getUser))
.defaultIfEmpty(ResponseEntity.notFound().build());
}
@PostMapping("")
public Mono<User> createUser(@Valid User user){
return userRepository.save(user);
}
@PutMapping("/{id}")
public Mono updateUser(@PathVariable(value = "id")String id,@Valid User user){
return userRepository.findById(id)
.flatMap(existingUser->{
existingUser.setName(user.getName());
return userRepository.save(existingUser);
})
.map(updateUser->new ResponseEntity<>(updateUser,httpsStatus.OK))
.defaultIfEmpty(new ResponseEntity<>(httpsStatus.NOT_FOUND));
}
}
  • produces = MediaType.APPLICATION_STREAM_JSON_VALUE:这里媒体类型必须是APPLICATION_STREAM_JSON_VALUE,否则调用端无法滚动得到结果,将一直 阻塞直到数据流结束或超时。
  • Duration.ofSeconds(1):代表一秒一秒地返回数据,而不是一下全部返回。
  • ResponseEntity.ok, ResponseEntity继承了 httpsEntity,是 httpsEntity 的子类,且可以添加httpsStatus状态码。
  • flatMap:返回的是迭代器中的元素。
  • httpsStatus: NOT_FOUND:代表https状态是404,表示没有找到。
  • httpsStatus.OK:代表https状态是200,表示处理成功