SpringBoot整合WebFlux、MongoDB实现增删改

WebFlux是异步的、非阻塞的响应式编程，目前支持的数据库有MongoDB和Redis，并不支持MySql，今天就整合MongoDB来实现一套增删改查，感受一下最终的效果。

首先，在pom中引入WebFlux及MongoDB的架包；

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
       <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
        </dependency>

第二步，在配置文件中，配置MongoDB的连接，需要在本地安装MongoDB数据库；

spring.data.mongodb.port=27017
spring.data.mongodb.host=localhost
#spring.data.mongodb.username=
#spring.data.mongodb.password=123456
spring.data.mongodb.database=test

第三步，controller、service、dao层代码；

import java.time.Duration;

import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import com.example.demo.service.UserService;
import com.example.demo.vo.User;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * controller层
 * @author 程就人生
 * @Date
 */
@Controller
@RequestMapping("/user")
public class UserController {
    
    @Autowired
    private UserService userService;
    
    /**
     * 列表页面
     * @return
     */
    @GetMapping("/userList")
    public String userList(final Model model) {
        //直接将列表渲染到页面
        //final Flux<User> userList = userService.findAll();
        //model.addAttribute("userList", userList);
        return "userList";
    }
    
    /**
     * 新增页面
     * @return
     */
    @GetMapping("/add")
    public String user() {
        return "user";
    }

    /**
     * 保存
     * @param user
     * @return
     */
    @PostMapping
    @ResponseBody
    public Mono<User> save(User user) {
        return userService.save(user);
    }

    /**
     * 通过用户名删除用户，有返回值
     * @param username
     * @return
     */
    @DeleteMapping("/{username}")
    @ResponseBody
    public Mono<Long> deleteByUsername(@PathVariable String username) {
        return userService.deleteByUsername(username);
    }
    
    /**
     * 通过id删除用户,没有返回值
     * @param id
     * @return
     */
    @GetMapping("/delete/{id}")
    @ResponseBody
    public Mono<Void> delete(@PathVariable("id") String id){
        return userService.findById(id).flatMap(x->
        userService.deleteById(x.getId())); 
    }

    /**
     * 通过用户名获取用户
     * Mono返回
     * @param username
     * @return
     */
    @GetMapping("/{username}")
    @ResponseBody
    public Mono<User> findByUsername(@PathVariable String username) {
        //查找不到时返回一个新的User对象
        return userService.findByUsername(username).map(x-> x)
                .defaultIfEmpty(new User());
    }

    /**
     * 第一种方式，数据延迟获取，一次只返回一条，可以减轻带宽压力
     * Flux，返回一次或多次
     * @return
     */
    @GetMapping(value="", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    @ResponseBody
    public Flux<User> findAll() {
        //每条数据之间延迟2秒
        return userService.findAll().delayElements(Duration.ofSeconds(2));
    }
    
    /**
     * 第二种方式，返回list<User>集合
     * Flux，返回一次或多次
     * @return
     */
    @GetMapping("/list")
    @ResponseBody
    public Flux<User> list(){
        Flux<User> flux = userService.findAll().map(x->{
            User user = new User();
            BeanUtils.copyProperties(x,user);
            return user;
        });
        return flux;
    }

}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.example.demo.dao.UserRepository;
import com.example.demo.vo.User;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
 * service层
 * @author 程就人生
 * @Date
 */
@Service
public class UserService {
    
    @Autowired
    private UserRepository userRepository;

    /**
     * 保存或更新。
     * 如果传入的user没有id属性，由于username是unique的，在重复的情况下有可能报错，
     * 这时找到以保存的user记录用传入的user更新它。
     */
    public Mono<User> save(User user) {
        return userRepository.save(user)
                .onErrorResume(e ->     // 进行错误处理
                        userRepository.findByUsername(user.getUsername())   // 找到username重复的记录
                                .flatMap(originalUser -> {      // 由于函数式为User -> Publisher，所以用flatMap
                                    user.setId(originalUser.getId());
                                    return userRepository.save(user);   // 拿到ID从而进行更新而不是创建
                                }));
    }

    public Mono<Long> deleteByUsername(String username) {
        return userRepository.deleteByUsername(username);
    }

    public Mono<User> findByUsername(String username) {
        return userRepository.findByUsername(username);
    }
    
    public Flux<User> findAll() {
        return userRepository.findAll();
    }
    
    public Mono<User> findById(String id){
        return userRepository.findById(id);
    }
    
    public Mono<Void> deleteById(String id) {
        return userRepository.deleteById(id);
    } 
}

import org.springframework.data.repository.reactive.ReactiveCrudRepository;

import com.example.demo.vo.User;

import reactor.core.publisher.Mono;

/**
 * Dao层
 * 继承ReactiveCrudRepository接口
 * @author 程就人生
 * @Date
 */
public interface UserRepository extends ReactiveCrudRepository<User, String> {
    
    //根据名称查找用户对象
    Mono<User> findByUsername(String username);
    
    //根据名称删除用户
    Mono<Long> deleteByUsername(String username);
}

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.Indexed;

import lombok.Data;

/**
 * user的实体类，使用lombok生成get、set方法
 * @author 程就人生
 * @Date
 */
@Data
public class User {
    
    @Id
    private String id;
    
    //对username做了唯一约束，需要在数据库里做索引
    @Indexed(unique = true)
    private String username;
    
    private String phone;
    
    private String email;

}

第四步，页面代码；
新增页面：

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="UTF-8"/>
    <title>用户新增</title>
</head>

<body>
<form class="m-t" role="form" id="userForm" th:action="@{/user}" method="post">
<table>
<tr>
    <td>username:</td>
    <td><input type="text" name="username" /></td>
</tr>
<tr>
    <td>phone:</td>
    <td><input type="text" name="phone" /></td>
</tr>
<tr>
    <td>email:</td>
    <td><input type="text" name="email" /></td>
</tr>
</table>
<button type="button" id="addUser" >保存</button>
</form>
</body>
<script src="https://cdn.staticfile.org/jquery/1.10.2/jquery.min.js" ></script>
<script th:inline="javascript" >
    $("#addUser").click(function(){
        $.ajax({
            type: "POST",
            url: "/user",
            data: $("#userForm").serialize(),
            dataType: "json",
            success : function(data) {
                if(data&&data.id){
                    alert("添加成功~！");
                }else{
                    alert("添加失败~！");
                }
                console.log("返回数据：" + data);
           }
       });
    });
</script>
</html>

列表页面：

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="UTF-8"/>
    <title>用户列表</title>
</head>

<body>
<div>
    <div id="dataModule"></div><br/>
    <div id="note" style="width: 100%;" ></div>
</div>
<script src="https://cdn.staticfile.org/jquery/1.10.2/jquery.min.js" ></script>
<script th:inline="javascript" >
//删除操作
function del(id){
    $.ajax({
        type: "get",
        url: "/user/delete/"+id,
        dataType: "json",
        success : function(data) {
            console.log("数据：" + data);  
       }
   });
}
//删除操作
function delByName(name){
    $.ajax({
        type: "DELETE",
        url: "/user/"+name,
        dataType: "json",
        success : function(data) {
            console.log("数据：" + data);
            if(data=='1'){
                alert("删除成功！");
            }else{
                alert("删除失败！");
            }
       }
   });
}
//根据名称获取对象
function getByName(name){
    $.ajax({
        type: "get",
        url: "/user/"+name,
        dataType: "json",
        success : function(data) {
            console.log("数据：" + data);  
            alert(data.username);
       }
   });
}
var time=1;
$(function() {  
    var xhr = new XMLHttpRequest();
    xhr.open('GET', '/user');
    //xhr.open('GET', '/user/list');
    xhr.send(null);//发送请求
    xhr.onreadystatechange = function() {
        console.log("s响应状态：" + xhr.readyState);
        //2是空响应，3是响应一部分，4是响应完成
        if (xhr.readyState > 2) {
            //这儿可以使用response(对应json)与responseText（对应text）
            var newData = xhr.response.substr(xhr.seenBytes);
            newData = newData.replace(/\n/g, "#");
            newData = newData.substring(0, newData.length - 1);         
            var data = newData.split("#");
            //newData = newData.substring(0, newData.length); //请求user/list时
            if(newData){
                console.log("返回数据：" + newData);
                //var data = JSON.parse(newData);
                //显示加载次数，和大小
                $("#dataModule").append("第"+time+"次数据响应"+data.length+"条<br/>");
                
                $("#note").append("<div style='clear: both;'>第"+time+"次数据响应"+data.length+"条</div><div id='note"+time+"' style='width: 100%;'></div>");
                var html="";
                console.log("数据：" + data);          
                for(var i=0;i<data.length;i++) {
                     var obj = JSON.parse(data[i]);
                     //var obj = data[i];
                     html=html + "<div style='margin-left: 10px;margin-top: 10px; width: 80px;height: 80px;background-color: gray;float: left;'>"+obj.username+","+obj.phone+ "," + obj.email;
                     html=html + "<label onclick='getByName(\""+obj.username+"\")' >获取</label>  <label onclick='del(\""+obj.id+"\")' >删除</label>  <label onclick='delByName(\""+obj.username+"\")' >删除1</label></div>"
                }           
                $("#note"+time).html(html);
                time++;
                xhr.seenBytes = xhr.response.length;
            }
        }
    }   
})
</script>
</body>
</html>

最后，测试；

新增测试
列表获取
单个获取
删除测试

总结
从使用WebFlux实现增删改查的结果来看，对于使用者来说，新增、修改、删除体验不到什么变化，在列表展示时，数据还是一次性从数据库中获取的，然后以较小的流量一点一点地反馈给前台。