前言

        从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