Статьи

Реактивные веб-приложения, использующие SpringWebFlux

Январь 13, 2019

1. Введение в реактивное программирование

Реактивное программирование — это термин, придуманный для приложений, который имеет следующие характеристики:

  • Неблокирующие приложения
  • Управляемый событиями и асинхронный
  • Требуется небольшое количество потоков для вертикального масштабирования (т. Е. В JVM)

Как и объектно-ориентированное программирование, функциональное программирование или процедурное программирование, реактивное программирование — это просто еще одна парадигма программирования. Это делает нашу программу: Responsive, Resilient, Elastic.

2. Реактивное программирование весной

Spring Framework использует Reactor для собственной реактивной поддержки. Reactor — это реализация Reactive Streams (издатели, представленные в Java9). Реактор имеет следующие два типа данных:

  • Поток (это поток, который может излучать 0 или более элементов)
  • Mono (это поток, который может излучать 0 или 1 элемент)

Spring предоставляет эти типы из своего API, что делает приложение реактивным.

В Spring 5 был представлен новый модуль под названием WebFlux , который предоставляет поддержку для создания реагирующих веб-приложений с использованием: HTTP (REST) ​​и веб-сокетов.

Spring Web Flux поддерживает следующие две модели:

  • Функциональная модель
  • Модель аннотации

В этой статье мы рассмотрим функциональную модель.

Следующая таблица сравнивает обычный Spring с Web Flux:

Традиционный стек Реактивный стек
Spring Web MVC Spring WebFlux
Отображение контроллера и обработчика Функции маршрутизатора
Сервлет API HTTP / Реактивный поток
Контейнеры для сервлетов Любой контейнер сервлетов с поддержкой Servlet 3.1+, Tomcat 8.x, Jetty, Netty, UnderTow

3. Вариант использования

REST API с использованием Spring Web Flux должен быть создан для системы управления сотрудниками, которая будет предоставлять CRUD для сотрудников.

Примечание: слой DAO проекта жестко закодирован.

4. Программное обеспечение и среда необходимы

  • Java: 1.8 или выше
  • Maven: 3.3.9 или выше
  • Затмение Луны или выше
  • Spring Boot: 2.0.0.M4
  • Spring Boot Starter WebFlux
  • Почтальон для тестирования приложения

5. Порядок подачи заявки

Функциональная модель Spring5 WebFlux является альтернативой использованию аннотаций в стиле Spring MVC. В функциональной модели Spring WebFlux маршрутизаторы и функции-обработчики используются для создания приложения MVC. Запрос @RequestMapping направляется через функцию маршрутизатора (альтернатива аннотациям, таким как @RequestMapping ), а запрос обрабатывается через функцию-обработчик (альтернативу @Controller обработчикам @Controller ).

Каждая функция-обработчик примет ServerRequest ( org.springframework.web.reactive.function.server.ServerRequest ) в качестве параметра и в результате вернет Mono<ServerResponse> или Flux<ServerResponse> ( org.springframework.web.reactive.function.server.ServerResponse ).

6. Код для варианта использования и описание

pom.xml

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.webflux</groupId>
    <artifactId>Demo_Spring_MVC_Web_Flux</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <repositories>
        <repository>
            <id>spring-snapshots</id>
            <name>Spring Snapshots</name>
            <url>https://repo.spring.io/snapshot</url>
            <snapshots>
                <enabled>true</enabled>
            </snapshots>
        </repository>
        <repository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>
    <pluginRepositories>
        <pluginRepository>
            <id>spring-snapshots</id>
            <name>Spring Snapshots</name>
            <url>https://repo.spring.io/snapshot</url>
            <snapshots>
                <enabled>true</enabled>
            </snapshots>
        </pluginRepository>
        <pluginRepository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </pluginRepository>
    </pluginRepositories>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.0.M4</version>
        <relativePath />
        <!-- lookup parent from repository -->
    </parent>
    <properties>
        <project.build.sourceEncoding>UTF-8
             </project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8
            </project.reporting.outputEncoding>
             <!-- Configuring Java 8 for the Project -->
        <java.version>1.8</java.version>
    </properties>
      <!--Excluding Embedded tomcat to make use of the Netty Server-->
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
               <exclusion>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-tomcat</artifactId>
               </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

EmployeeDAO.java

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
package com.webflux.dao;
import java.util.LinkedHashMap;
import java.util.Map;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import com.webflux.bussiness.bean.Employee;
@Repository
public class EmployeeDAO {
    /**
     * Map is used to Replace the Database
     * */
    static public Map mapOfEmloyeess = new LinkedHashMap();
    static int count=10004;
    static
    {
       mapOfEmloyeess.put(10001, new Employee("Jack",10001,12345.6,1001));
       mapOfEmloyeess.put(10002, new Employee("Justin",10002,12355.6,1002));
       mapOfEmloyeess.put(10003, new Employee("Eric",10003,12445.6,1003));
    }
 
    /**
     * Returns all the Existing Employees as Flux
     * */
    public Flux getAllEmployee(){
        return Flux.fromStream(mapOfEmloyeess.values().stream());      
    }
 
    /**Get Employee details using EmployeeId .
     * Returns a Mono response with Data if Employee is Found
     * Else returns a null
     * */
    public Mono getEmployeeDetailsById(int id){
         
        Monores = null;
        Employee emp =mapOfEmloyeess.get(id);
        if(emp!=null){
            res=Mono.just(emp);
        }
        return res;
    }
 
    /**Create Employee details.
     * Returns a Mono response with auto-generated Id
     * */
    public Mono addEmployee(Employee employee){
        count++;
        employee.setEmployeeId(count);
        mapOfEmloyeess.put(count, employee);
        return Mono.just(count);
    }
 
    /**Update the Employee details,
     * Receives the Employee Object and returns the updated Details
     * as Mono 
     * */
    public Mono updateEmployee (Employee employee){
        mapOfEmloyeess.put(employee.getEmployeeId(), employee);
        return Mono.just(employee);
    }
 
    /**Delete the Employee details,
     * Receives the EmployeeID and returns the deleted employee Details
     * as Mono 
     * */
    public Mono removeEmployee (int id){
        Employee emp= mapOfEmloyeess.remove(id);
        return Mono.just(emp);
    }  
}

Можно заметить, что все методы EmployeeDAO возвращают ответ Mono или Flux, что делает уровень DAO реактивным.

EmployeeHandler.java

001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071
072
073
074
075
076
077
078
079
080
081
082
083
084
085
086
087
088
089
090
091
092
093
094
095
096
097
098
099
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package com.webflux.web.handler;
import static org.springframework.web.reactive.function.BodyInserters.fromObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.web.reactive.function.BodyExtractors;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import com.webflux.bussiness.bean.Employee;
import com.webflux.dao.EmployeeDAO;
@Controller
public class EmployeeHandler {
 
    @Autowired
    private EmployeeDAO employeeDAO;
     
     
    /**
     * Receives a ServerRequest.
     * Invokes the method getAllEmployee() from EmployeeDAO.
     * Prepares a Mono and returns the same.
     * */  
      public Mono getEmployeeDetails(ServerRequest request) {
        Flux  res=employeeDAO.getAllEmployee();    
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
        .body(res,Employee.class);
    }
     
    /**
     * Receives a ServerRequest.
     * Extracts the Path Variable (named id) from the Request.
     * Invokes the method [getEmployeeDetailsById()] from EmployeeDAO.
     * Verifies if the object returned in the previous step is null
     * then returns a Bad request with appropriate message.
     * Else Returns the Mono with the Employee Data.
     * */
    public Mono getEmployeeDetailByEmployeeId(ServerRequest request) {
        //Extracts the Path Variable id from the Request
        int id =Integer.parseInt(request.pathVariable("id"));
        Mono employee = employeeDAO.getEmployeeDetailsById(id);
        Mono res= null;
        if(employee==null){
            res=ServerResponse.badRequest().body
                                    (fromObject("Please give a valid employee Id"));
        }
        else{
            //Converting Mono of Mono type to Mono
            res=employee.flatMap(x->ServerResponse.ok().body(fromObject(x)));
        }
        return res;
    }
 
    /**
     * Receives a ServerRequest.
     * Makes use of BodyExtractors and Extracts the Employee Data as
     * Mono from the ServerRequest.
     * Invokes the method [addEmployee()] of the EmployeeDAO.
     * Prepares a Mono and returns the same.
     * */
    public Mono addEmployee(ServerRequest request) {
        Mono requestBodyMono = request.body(BodyExtractors.toMono(Employee.class));
            Mono mono= employeeDAO.addEmployee(requestBodyMono.block());
        //Converting Mono of Mono type to Mono
        Mono res= mono.flatMap(x->ServerResponse.ok().body
                                    (fromObject("Employee Created with Id"+x)));
        return res;
    }
 
     
    /**
     * Receives a ServerRequest.
     * Makes use of BodyExtractors and Extracts the Employee Data as
     * Mono from the ServerRequest.
     * Finds the Employee and updates the details by invoking updateEmployee() of  
     * EmployeeDAO.
     * Prepares a Mono and returns the same.
     * */
    public Mono updateEmployee(ServerRequest request) {
        Mono requestBodyMono = request.body(BodyExtractors.toMono(Employee.class));
        Employee employee =  requestBodyMono.block();
          Mono employeeRet = employeeDAO.getEmployeeDetailsById(employee.getEmployeeId());
          Mono res= null;
        if(employeeRet==null){
            res=ServerResponse.badRequest().body(fromObject
                            ("Please Give valid employee details to update"));
        }
 
        else{
            Mono emp= employeeDAO.updateEmployee(employee);
             
            //Converting Mono of Mono type to Mono
            res=emp.flatMap(x->ServerResponse.ok().body(fromObject(x)));
        }
        return res;            
    }
 
    /**
     * Receives a ServerRequest.
     * Makes use of BodyExtractors and Extracts the Employee Data as
     * Mono from the ServerRequest.
     * Finds the Employee and deletes the details by invoking removeEmployee() of  
     * EmployeeDAO.
     * Prepares a Mono and returns the same.
     * */
    public Mono deleteEmployee(ServerRequest request) {
        int myId = Integer.parseInt(request.pathVariable("id"));
        Mono res= null;
        if (employeeDAO.getEmployeeDetailsById(myId) == null) {
            res=ServerResponse.badRequest().body
                      (fromObject("Please Give valid employee details to delete"));
        }else{
          Mono employee = employeeDAO.removeEmployee(myId);
         
              //Converting Mono of Mono type to Mono
          res=employee.flatMap(x->ServerResponse.ok().body(fromObject(x)));
           
        }
        return res;
    }
}

Можно заметить, что все методы обработчика возвращают Mono<ServerResponse> , тем самым делая слой представления реактивным.

Примечание . Метод обработчика событий должен принимать запрос ServerRequest и возвращать Mono<ServerResponse>

RouterConfiguration.java

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
package com.webflux.web.router.config;
import static org.springframework.web.reactive.function.server.RequestPredicates.DELETE;
import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RequestPredicates.POST;
import static org.springframework.web.reactive.function.server.RequestPredicates.PUT;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;
import com.webflux.web.handler.EmployeeHandler;
@Configuration
/**
 * Router is configuration class.
 * It links the incoming requests with appropriate HTTP methods to the
 * respective method of the EmployeeHandler.
 * Method references are used for the mapping.
 * */
public class RouterConfiguration{
    @Autowired
    EmployeeHandler employeeHandler;
     
    @Bean
    public RouterFunction monoRouterFunction() {
     RouterFunctionrouterFunction=  RouterFunctions.       
              route(GET("/emp/controller/getDetails").
              and(accept(MediaType.APPLICATION_JSON)),         
              employeeHandler::getEmployeeDetails)
              
             .andRoute(GET("/emp/controller/getDetailsById/{id}")
             .and(accept(MediaType.APPLICATION_JSON)),           
              employeeHandler::getEmployeeDetailByEmployeeId)
     
             .andRoute(POST("/emp/controller/addEmp")
             .and(accept(MediaType.APPLICATION_JSON)),
              employeeHandler::addEmployee)
                         
             .andRoute(PUT("/emp/controller/updateEmp")
             .and(accept(MediaType.APPLICATION_JSON)),
             employeeHandler::updateEmployee)
                         
             .andRoute(DELETE("/emp/controller/deleteEmp/{id}")
             .and(accept(MediaType.APPLICATION_JSON)),
             employeeHandler::deleteEmployee);
        return routerFunction;
    }
   
}

ApplicationBootUp.java

01
02
03
04
05
06
07
08
09
10
package com.webflux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ApplicationBootUp {
    public static void main(String[] args) {
        SpringApplication.run(ApplicationBootUp.class);
    }
 
}

Внутри application.properties указан только порт сервера: server.port=8090 .

Приложение может быть развернуто с помощью команды: clean install spring-boot:run и протестируйте с помощью postman client .

7. Ссылки:

8. Скачать проект Eclipse

Скачать
Вы можете скачать полный исходный код этого примера здесь: SpringWebFlux
47
Share: