前言 从spring5中加入webflux的消息现世已经有一段时间了，也发现了一些朋友的公司在使用webfux，但是大多数都是用作为服务器之间的一些通讯、路由控制来使用，然而真正的把他当着一个we
前言
从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;@Servicepublic 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数据更改不执行代码：
@Overridepublic 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)）的结果的，都是数据改动不成功的，不知道为什么....
