WebFlux拨云见日之前端整合，悟了吗？

浏览数:1,512

2019-7-4

前言

从spring5中加入webflux的消息现世已经有一段时间了，也发现了一些朋友的公司在使用webfux，但是大多数都是用作为服务器之间的一些通讯、路由控制来使用，然而真正的把他当着一个web服务来提供给前端API的并木有。早年我就接触了bigpipe的概率了，但是在java的领域并不怎么活，单流的数据响应是否好像类似呢？于是我就研究了webflux和前端整合分享一下大家共同探讨…

WebFlux

WebFlux 模块的名称是 spring-webflux，名称中的 Flux 来源于 Reactor 中的类 Flux。Spring webflux 有一个全新的非堵塞的函数式 Reactive Web 框架，可以用来构建异步的、非堵塞的、事件驱动的服务，在伸缩性方面表现非常好.

实战效果展示

第一处：

是对推送SSE API允许网页获得来自服务器的更新（HTML5），用于创建到服务器的单向连接，服务器通过这个连接可以发送任意数量的数据。服务器响应的MIME类型必须是text/event-stream，而且是浏览器中的JavaScript API 能解析格式输出。SSE 支持短轮询、长轮询和HTTP 流，而且能在断开连接时自动确定何时重新连接。

java代码：

@RestController
@RequestMapping("/sse")
public class SseController {
 
    private int count_down_sec=3*60*60;
     
    @GetMapping(value="/countDown",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Object>> countDown() {
         
        return Flux.interval(Duration.ofSeconds(1))
            .map(seq -> Tuples.of(seq, getCountDownSec()))
            .map(data -> ServerSentEvent.<Object>builder()
                    .event("countDown")
                    .id(Long.toString(data.getT1()))
                    .data(data.getT2().toString())
                    .build());
    }
     
    private String getCountDownSec() {
        if (count_down_sec>0) {
            int h = count_down_sec/(60*60);
            int m = (count_down_sec%(60*60))/60;
            int s = (count_down_sec%(60*60))%60;
            count_down_sec--;
            return "活动倒计时："+h+" 小时 "+m+" 分钟 "+s+" 秒";
        }
        return "活动倒计时：0 小时 0 分钟 0 秒";
    }
}

HTML代码：

//js代码
<script>
    //记录加载次数
    var time=1;
    if (typeof (EventSource) !== "undefined") {
        var source = new EventSource("/sse/countDown");
        console.log(source);
        source.addEventListener("countDown", function(e) {
            document.getElementById("result").innerHTML = e.data;
        }, false);//使用false表示在冒泡阶段处理事件，而不是捕获阶段。
    } else {
        document.getElementById("result").innerHTML = "抱歉，你的浏览器不支持 server-sent 事件...";
    }
</script>
//html代码
<div id="result"></div><br/>

webflux中的sse并不是什么新的东西，在之前都有出现过，就不多介绍了。

第二三处就是对webflux中的Flux接口信息

java代码(主要是针对Mongo)

Entity：

@Data
@EqualsAndHashCode(callSuper=false)
public class Commodity extends ParentEntity implements Serializable{
 
    private static final long serialVersionUID = 6659341305838439447L;
     
    /**
     *  店铺ID 
     */
    private Long shopId;
     
    /**
     *  商品名
     */
    @NotNull(message = "商品名不能为空")
    @Pattern(regexp = "^.{0,50}$",message = "商品名必须是小于50位字符串")
    private String name;
     
    /**
     *  商品详情
     */
    @Pattern(regexp = "^.{0,500}$",message = "商品详情必须是小于500位字符串")
    private String details;
     
    /**
     *  商品图片地址
     */
    private String imageUrl;
     
    /**
     *  商品图片地址
     */
    private Integer type;
     
    /**
     *  商品单价
     */
    @Min(value = 0)
    private BigDecimal price;
     
    /**
     *  商品库存
     */
    @Min(value = 0)
    private Integer pcs;
}

Entity的父类：

public class ParentEntity {
    /**
     *  商品ID 
     */
    public Long id;
 
    public Long getId() {
        return id;
    }
 
    public void setId(Long id) {
        this.id = id;
    }
     
}

DAO:

import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import com.flying.cattle.wf.entity.Commodity;
 
public interface CommodityRepository extends ReactiveMongoRepository<Commodity, Long>{
 
}

SERVICE:

import com.flying.cattle.wf.aid.IService;
import com.flying.cattle.wf.entity.Commodity;
 
public interface CommodityService extends IService<Commodity>{
     
}

SERVICE的父类：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
 
/**
 * <p>
 * 顶级 Service
 * </p>
 *
 * @author BianPeng
 * @since 2019/6/17
 */
public interface IService<T> {
     
    /**
     * <p>
     * 插入一条记录（选择字段，策略插入）
     * </p>
     *
     * @param entity 实体对象
     */
    Mono<T> insert(T entity);
     
    /**
     * <p>
     * 根据 ID 删除
     * </p>
     * @param id 主键ID
     */
    Mono<Void> deleteById(Long id);
     
    /**
     * <p>
     * 根据 ID 删除
     * </p>
     * @param id 主键ID
     */
    Mono<Void> delete(T entity);
     
    /**
     * <p>
     * 根据 ID 选择修改
     * </p>
     * @param entity 实体对象
     */
    Mono<T> updateById(T entity);
     
    /**
     * <p>
     * 根据 ID 查询
     * </p>
     * @param id 主键ID
     */
    Mono<T> findById(Long id);
     
    /**
     * <p>
     * 查询所有
     * </p>
     */
    Flux<T> findAll();
}

SERVICE-IMPL

import org.springframework.stereotype.Service;
import com.flying.cattle.wf.aid.ServiceImpl;
import com.flying.cattle.wf.dao.CommodityRepository;
import com.flying.cattle.wf.entity.Commodity;
import com.flying.cattle.wf.service.CommodityService;
 
@Service
public class CommodityServiceImpl extends ServiceImpl<CommodityRepository, Commodity> implements CommodityService {
 
}

SERVICE-IMPL 父类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
 
/**
 * <p>
 * IService 实现类（ 泛型：M 是 mapper 对象，T 是实体 ， PK 是主键泛型 ）
 * </p>
 *
 * @author BianPeng
 * @since 2019/6/17
 */
public class ServiceImpl<M extends ReactiveMongoRepository<T,Long>, T> implements IService<T> {
 
    @Autowired
    protected M baseDao;
     
    @Override
    public Mono<T> insert(T entity) {
        return baseDao.save(entity);
    }
 
    @Override
    public Mono<Void> deleteById(Long id) {
        // TODO Auto-generated method stub
        return baseDao.deleteById(id);
    }
 
    @Override
    public Mono<Void> delete(T entity) {
        // TODO Auto-generated method stub
        return baseDao.delete(entity);
    }
 
    @Override
    public Mono<T> updateById(T entity) {
        // TODO Auto-generated method stub
        return baseDao.save(entity);
    }
 
    @Override
    public Mono<T> findById(Long id) {
        // TODO Auto-generated method stub
        return baseDao.findById(id);
    }
 
    @Override
    public Flux<T> findAll() {
        // TODO Auto-generated method stub
        return baseDao.findAll();
    }
 
}

CONTROLLER：

import java.math.BigDecimal;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.flying.cattle.wf.aid.AbstractController;
import com.flying.cattle.wf.entity.Commodity;
import com.flying.cattle.wf.service.CommodityService;
import com.flying.cattle.wf.utils.ValidationResult;
import com.flying.cattle.wf.utils.ValidationUtils;
import reactor.core.publisher.Mono;
 
@RestController
@RequestMapping("/commodity")
public class CommodityController extends AbstractController<CommodityService, Commodity> {
 
    Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @GetMapping("/add")
    public Mono<Commodity> add() throws Exception {
        Commodity obj = new Commodity();
        Long id=super.getAutoIncrementId(obj);
        obj.setId(id);
        obj.setShopId(1l);
        obj.setName("第" + id + "个商品");
        obj.setDetails("流式商品展示");
        obj.setImageUrl("/aa/aa.png");
        obj.setPcs(1);
        obj.setPrice(new BigDecimal(998));
        obj.setType(1);
 
        ValidationResult vr = ValidationUtils.validateEntity(obj);
        if (!vr.isHasErrors()) {
            return baseService.insert(obj);
        } else {
            throw new Exception(vr.getFirstErrors());
        }
    }
}

CONTROLLER父类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
 
import com.flying.cattle.wf.entity.ParentEntity;
import com.flying.cattle.wf.service.impl.RedisGenerateId;
 
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
 
/**   
 * <p>自动生成工具：mybatis-dsc-generator</p>
 * 
 * <p>说明： 资金归集API接口层</P>
 * @version: V1.0
 * @author: BianPeng
 *
 */
public class AbstractController<S extends IService<T>,T extends ParentEntity>{
     
    @Autowired
    private RedisGenerateId redisGenerateId;
 
    @Autowired
    protected S baseService;
     
    @GetMapping("/getId")
    public Long getAutoIncrementId(T entity){
        return redisGenerateId.generate(entity.getClass().getName());
    }
     
    @PostMapping("/save")
    public Mono<T> save(T entity) {
        entity.setId(getAutoIncrementId(entity));
        return baseService.insert(entity);
    }
 
    @PostMapping("/deleteById")
    public Mono<String> deleteById(Long id) {
        // TODO Auto-generated method stub
        return baseService.deleteById(id)
                .then(Mono.just("ok"))
                .defaultIfEmpty("not found");
    }
 
    @PostMapping("/delete")
    public Mono<String> delete(T entity) {
        // TODO Auto-generated method stub
        return baseService.delete(entity)
                .then(Mono.just("ok"))
                .defaultIfEmpty("not found");
    }
 
    @PostMapping("/updateById")
    public Mono<T> updateById(T entity) {
        // TODO Auto-generated method stub
        return baseService.updateById(entity);
    }
 
    @GetMapping("/getById/{id}")
    public Mono<T> findById(@PathVariable("id") Long id) {
        // TODO Auto-generated method stub
        return baseService.findById(id);
    }
 
    @GetMapping(value="/findAll",produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<T> findAll() {
        // TODO Auto-generated method stub
        return baseService.findAll();
    }
}

HTML代码：

<html>
<head>
<meta charset="UTF-8">
<title>商品展示</title>
<script src="/js/jquery-2.1.1.min.js"></script>
<script>
    //记录加载次数
    var time=1;
    if (typeof (EventSource) !== "undefined") {
        var source = new EventSource("/sse/countDown");
        console.log(source);
        source.addEventListener("countDown", function(e) {
            document.getElementById("result").innerHTML = e.data;
        }, false);//使用false表示在冒泡阶段处理事件，而不是捕获阶段。
    } else {
        document.getElementById("result").innerHTML = "抱歉，你的浏览器不支持 server-sent 事件...";
    }
    /************************以上是SSE的JS************************/
    $(function() {
        var xhr = new XMLHttpRequest();
        xhr.open('GET', '/commodity/findAll');
        xhr.send(null);//发送请求
        xhr.onreadystatechange = function() {
            //2是空响应，3是响应一部分，4是响应完成
            if (xhr.readyState > 2) {
                //这儿可以使用response与responseText，因为我接口返回的就是String数据，所以选择responseText
                var newData = xhr.response.substr(xhr.seenBytes);
                newData = newData.replace(/\n/g, "#");
                newData = newData.substring(0, newData.length - 1);
                var data = newData.split("#");
                //显示加载次数，和大小
                $("#dataModule").append("第"+time+"次数据响应"+data.length+"条          ");
                 
                $("#note").append("<div style='clear: both;'>第"+time+"次数据响应"+data.length+"条</div><div id='note"+time+"' style='width: 100%;'></div>");
                var html="";
                for (var i = 0; i < data.length; i++) {
                     var obj = JSON.parse(data[i])
                     html=html + "<div style='margin-left: 10px;margin-top: 10px; width: 80px;height: 80px;background-color: gray;float: left;'>"+obj.name+"</div>";
                }
                $("#note"+time).html(html);
                time++;
                xhr.seenBytes = xhr.response.length;
            }
        }
    })
</script>
</head>
<body>
    <div id="result"></div><br/>
    <div id="dataModule"></div><br/>
    <div id="note" style="width: 100%;" >
    </div>
</body>
</html>

前端的代码就这个样子，展示出来的数据就是开始那张图，数据反应多少条前端马上就展示数据，不需要等待接口反应完成并接收完所以数据才展示数据。现在webflux的使用其实还并不是太普及，很多东西都得需要我自己去挖掘，有兴趣的朋友可以加群：340697945大家一起交流，相互学习。

遇见的问题：

serviceImpl数据更改不执行代码：

@Override
public Mono<Boolean> deleteById(Long id) {
    // TODO Auto-generated method stub
    baseDao.deleteById(id);
    return Mono.create(entityMonoSink -> entityMonoSink.success());
}

controller中不执行的代码（如果删除是源码中的方法）

@PostMapping("/deleteById")
public Mono<Boolean> deleteById(Long id) {
    // TODO Auto-generated method stub
    baseService.deleteById(id)
    return Mono.create(entityMonoSink -> entityMonoSink.success());      
}

我在执行的时候，发现删除不了数据，但是也没报错。最后发现只要controller中没直接返回ReactiveMongoRepository中方法（如baseDao.deleteById(id)或baseDao.save(entity)）的结果的，都是数据改动不成功的，不知道为什么….

源码地址：https://gitee.com/flying-cattle/infrastructure/tree/master/webFluxTest

作者：边鹏_尛爺鑫

原文地址：https://my.oschina.net/bianxin/blog/3063713