flux服务器推消息,SpringBoot整合WebFlux实现SSE事件

碧海云天97

2514人浏览 · 2021-08-12 10:16:17

碧海云天97 · 2021-08-12 10:16:17 发布

前言

在前台页面需要不停获取服务器端的数据时，无非有两种操作，一种是通过前台页面使用轮询的方式，定时向服务器后台发送请求，以获取最新的数据；另一种就是在前台页面和后台服务之间建立长连接，服务器端一有数据产生就向前端页面推送。

这里的SSE是服务器发送事件(Server-Sent Events) 的缩写，在WebFlux框架里，服务器端是如何向前端(或调用端)实现服务器发送事件的呢？在有前端页面的情况下，又是如何实现的呢？

带着上面的这些疑问，来了解WebFlux框架，WebFlux框架是一款响应式编程web框架，什么是响应式编程呢，根据wikipedia上的定义：

响应式编程是就是对于数据流和传播改变的一种声明式的编程规范。这意味着可以通过编程语言轻松地表达静态(例如数组)或动态(例如事件发射器)数据流，并且存在相关执行模型内的推断依赖性，这有助于自动传播数据流涉及的变化。

围绕着WebFlux框架的，有这么几个关键字，异步的、非阻塞的、响应式的，那么是不是能够实现数据一有变化，就通知到对应的调用端呢，这些还有待证实。

基于WebFlux框架的SSE应用

首先，在pom文件中，引入webflux框架；

org.springframework.boot

spring-boot-starter-webflux

org.springframework.boot

spring-boot-starter-thymeleaf

第二，html代码，共有四个页面；

sse.html页面代码：

服务器推送事件

//服务器推送事件

if (typeof (EventSource) !== "undefined") {

//第一种写法

//接收服务器倒计时时间推送，使用HTML5 服务器发送事件(Server-Sent Events)，参考资料：https://www.runoob.com/html/html5-serversentevents.html

var source = new EventSource("/sse/countDown");

console.log(source);

source.addEventListener("countDown", function(e) {

document.getElementById("result").innerHTML = e.data;

}, false);//使用false表示在冒泡阶段处理事件，而不是捕获阶段。

//第二种写法

//随机数获取

var source1 = new EventSource("/sse/retrieve");

//当抓取到消息时

source1.onmessage = function (evt) {

document.getElementById("data").innerHTML = "股票行情：" + evt.data;

};

} else {

//注意：ie浏览器不支持

document.getElementById("result").innerHTML = "抱歉，你的浏览器不支持 server-sent 事件...";

var xhr;

var xhr2;

if (window.XMLHttpRequest){

//IE7+, Firefox, Chrome, Opera, Safari浏览器支持该方法

xhr=new XMLHttpRequest();

xhr2=new XMLHttpRequest();

}else{

//IE6, IE5 浏览器不支持，使用ActiveXObject方法代替

xhr=new ActiveXObject("Microsoft.XMLHTTP");

xhr2=new ActiveXObject("Microsoft.XMLHTTP");

}

console.log(xhr);

console.log(xhr2);

xhr.open('GET', '/sse/countDown');

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("#");

console.log("获取到的数据：" + data);

document.getElementById("result").innerHTML = data;

//长度重新赋值,下次截取时需要使用

xhr.seenBytes = xhr.response.length;

}

xhr2.open('GET', '/sse/retrieve');

xhr2.send(null);//发送请求

xhr2.onreadystatechange = function() {

console.log("s响应状态：" + xhr2.readyState);

//0: 请求未初始化，2 请求已接收，3 请求处理中，4 请求已完成，且响应已就绪

if (xhr2.readyState > 2) {

//这儿可以使用response(对应json)与responseText(对应text)

var newData1 = xhr2.response.substr(xhr2.seenBytes);

newData1 = newData1.replace(/\n/g, "#");

newData1 = newData1.substring(0, newData1.length - 1);

var data1 = newData1.split("#");

console.log("获取到的数据：" + data1);

document.getElementById("data").innerHTML = data1;

//长度重新赋值,下次截取时需要使用

xhr2.seenBytes = xhr2.response.length;

}

sse2.html页面代码：

服务器推送

$(function() {

var time=1;

var xhr;

if (window.XMLHttpRequest){

//IE7+, Firefox, Chrome, Opera, Safari浏览器支持该方法

xhr=new XMLHttpRequest();

}else{

//IE6, IE5 浏览器不支持，使用ActiveXObject方法代替

xhr=new ActiveXObject("Microsoft.XMLHTTP");

}

console.log(xhr);

xhr.open('GET', '/quotes');

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("#");

//显示加载次数，和大小

$("#dataModule").append("第"+time+"次数据响应"+data.length+"条
");

$("#note").append("

第"+time+"次数据响应"+data.length+"条

var html="";

console.log("数据：" + data);

for(var i=0;i

var obj = JSON.parse(data[i]);

html=html + "

"+obj.ticker+"

}

$("#note"+time).html(html);

time++;

//长度重新赋值,下次截取时需要使用

xhr.seenBytes = xhr.response.length;

}

})

sse3.html页面代码：

服务器推送

$(function() {

var time=1;

var xhr;

if (window.XMLHttpRequest){

//IE7+, Firefox, Chrome, Opera, Safari浏览器支持该方法

xhr=new XMLHttpRequest();

}else{

//IE6, IE5 浏览器不支持，使用ActiveXObject方法代替

xhr=new ActiveXObject("Microsoft.XMLHTTP");

}

console.log(xhr);

xhr.open('POST', '/quotes');

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);

console.log("数据：" + newData);

if(newData){

//将字符串类型的json转成json对象

var data = JSON.parse(newData.split("#"));

//显示加载次数，和大小

$("#dataModule").append("第"+time+"次数据响应"+data.length+"条
");

$("#note").append("

第"+time+"次数据响应"+data.length+"条

var html="";

console.log("数据：" + data);

for(var i=0;i

var obj = data[i];

html=html + "

"+obj.ticker+"

}

$("#note"+time).html(html);

time++;

//长度重新赋值,下次截取时需要使用

xhr.seenBytes = xhr.response.length;

}else{

$("#dataModule").append("响应完成！！！");

}

})

sse4.html页面代码：

服务器推送

$(function() {

var time=1;

var xhr;

if (window.XMLHttpRequest){

//IE7+, Firefox, Chrome, Opera, Safari浏览器支持该方法

xhr=new XMLHttpRequest();

}else{

//IE6, IE5 浏览器不支持，使用ActiveXObject方法代替

xhr=new ActiveXObject("Microsoft.XMLHTTP");

}

console.log(xhr);

xhr.open('POST', '/echo1');

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("#");

console.log("获取到的数据：" + data);

document.getElementById("dataModule").innerHTML = data;

}

})

注意：在前端页面，接收服务器的推送请求，需要html5的SSE支持，除了IE外，其他的浏览器都支持；

第三，后台代码；

import java.math.BigDecimal;

import java.math.MathContext;

import java.time.Instant;

/**

* 需要推送的实体类

* @author 程就人生

* @Date

public class Quote {

private static final MathContext MATH_CONTEXT = new MathContext(2);

private String ticker;

private BigDecimal price;

private Instant instant;

public Quote() {

}

public Quote(String ticker, BigDecimal price) {

this.ticker = ticker;

this.price = price;

}

public Quote(String ticker, Double price) {

this(ticker, new BigDecimal(price, MATH_CONTEXT));

}

@Override

public String toString() {

return "Quote{" +

"ticker='" + ticker + '\'' +

", price=" + price +

", instant=" + instant +

'}';

}

public final String getTicker() {

return ticker;

}

public final void setTicker(String ticker) {

this.ticker = ticker;

}

public final BigDecimal getPrice() {

return price;

}

public final void setPrice(BigDecimal price) {

this.price = price;

}

public final Instant getInstant() {

return instant;

}

public final void setInstant(Instant instant) {

this.instant = instant;

}

import java.math.BigDecimal;

import java.math.MathContext;

import java.time.Duration;

import java.time.Instant;

import java.util.ArrayList;

import java.util.List;

import java.util.Random;

import java.util.stream.Collectors;

import org.springframework.stereotype.Component;

import com.example.entity.Quote;

import reactor.core.publisher.Flux;

/**

* 推送数据，模拟生成

* @author 程就人生

* @Date

@Component

public class QuoteGenerator {

private final MathContext mathContext = new MathContext(2);

private final Random random = new Random();

private final List prices = new ArrayList<>();

/**

* 生成行情数据

public QuoteGenerator() {

this.prices.add(new Quote("CTXS", 82.26));

this.prices.add(new Quote("DELL", 63.74));

this.prices.add(new Quote("GOOG", 847.24));

this.prices.add(new Quote("MSFT", 65.11));

this.prices.add(new Quote("ORCL", 45.71));

this.prices.add(new Quote("RHT", 84.29));

this.prices.add(new Quote("VMW", 92.21));

}

public Flux fetchQuoteStream(Duration period) {

// 需要周期生成值并返回，使用 Flux.interval

return Flux.interval(period)

// In case of back-pressure, drop events

.onBackpressureDrop()

// For each tick, generate a list of quotes

.map(this::generateQuotes)

// "flatten" that List into a Flux

.flatMapIterable(quotes -> quotes)

.log("io.spring.workshop.stockquotes");//以日志的形式输出

}

/**

* Create quotes for all tickers at a single instant.

private List generateQuotes(long interval) {

final Instant instant = Instant.now();

return prices.stream()

.map(baseQuote -> {

BigDecimal priceChange = baseQuote.getPrice()

.multiply(new BigDecimal(0.05 * this.random.nextDouble()), this.mathContext);

Quote result = new Quote(baseQuote.getTicker(), baseQuote.getPrice().add(priceChange));

result.setInstant(instant);

return result;

})

.collect(Collectors.toList());

}

import java.time.Duration;

import org.springframework.http.MediaType;

import org.springframework.stereotype.Component;

import org.springframework.web.reactive.function.BodyInserters;

import org.springframework.web.reactive.function.server.ServerRequest;

import org.springframework.web.reactive.function.server.ServerResponse;

import com.example.entity.Quote;

import com.example.generator.QuoteGenerator;

import reactor.core.publisher.Flux;

import reactor.core.publisher.Mono;

/**

* 数据处理handler，相当于service层

* @author 程就人生

* @Date

@Component

public class QuoteHandler {

private final Flux quoteStream;

public QuoteHandler(QuoteGenerator quoteGenerator) {

this.quoteStream = quoteGenerator.fetchQuoteStream(Duration.ofMillis(1000 * 10)).share();

}

public Mono hello(ServerRequest request) {

Long start = System.currentTimeMillis();

return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)

.body(BodyInserters.fromObject("Hello Spring!" + start));

}

public Mono echo(ServerRequest request) {

return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)

.body(request.bodyToMono(String.class), String.class);

}

public Mono streamQuotes(ServerRequest request) {

Long start = System.currentTimeMillis();

System.out.println("--------------" + start + "--------------");

return ServerResponse.ok()

.contentType(MediaType.APPLICATION_STREAM_JSON) //返回多次

.body(this.quoteStream, Quote.class);

}

public Mono fetchQuotes(ServerRequest request) {

int size = Integer.parseInt(request.queryParam("size").orElse("10"));

return ServerResponse.ok()

.contentType(MediaType.APPLICATION_JSON) //返回一次

.body(this.quoteStream.take(size), Quote.class);

}

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.http.MediaType;

import org.springframework.web.reactive.function.server.RequestPredicates;

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.example.handler.QuoteHandler;

/**

* 路由，相当于Controller层

* @author 程就人生

* @Date

@Configuration

public class QuoteRouter {

@Bean

public RouterFunction route(QuoteHandler quoteHandler) {

return RouterFunctions

.route(RequestPredicates.GET("/hello1").and(RequestPredicates.accept(MediaType.TEXT_PLAIN)), quoteHandler::hello)

.andRoute(RequestPredicates.POST("/echo1").and(RequestPredicates.accept(MediaType.TEXT_PLAIN).and(RequestPredicates.contentType(MediaType.TEXT_PLAIN))), quoteHandler::echo)

//响应一次

.andRoute(RequestPredicates.POST("/quotes").and(RequestPredicates.accept(MediaType.APPLICATION_JSON)), quoteHandler::fetchQuotes)

//响应多次

.andRoute(RequestPredicates.GET("/quotes").and(RequestPredicates.accept(MediaType.APPLICATION_STREAM_JSON)), quoteHandler::streamQuotes);

}

import java.time.Duration;

import org.springframework.http.MediaType;

import org.springframework.http.codec.ServerSentEvent;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.ResponseBody;

import reactor.core.publisher.Flux;

import reactor.util.function.Tuples;

/**

* 服务器发送事件SSE(Server-Sent Events)

* 页面渲染及请求

* @author 程就人生

* @Date

@Controller

@RequestMapping("/sse")

public class SseController {

//三分钟倒计时

private int count_down_sec=3*60*60;

/**

* 推送页面1

* @return

@GetMapping

public String sse(){

return "sse";

}

/**

* 推送页面2

* @return

@GetMapping("/two")

public String sse2(){

return "sse2";

}

/**

* 推送页面3

* @return

@GetMapping("/three")

public String sse3(){

return "sse3";

}

/**

* 推送页面4

* @return

@GetMapping("/four")

public String sse4(){

return "sse4";

}

//报头设置为 "text/event-stream"，以便于发送事件流

@GetMapping(value="/countDown",produces = MediaType.TEXT_EVENT_STREAM_VALUE)

@ResponseBody

public Flux> countDown() {

//每一秒钟推送一次

return Flux.interval(Duration.ofSeconds(1))

.map(seq -> Tuples.of(seq, getCountDownSec()))

.map(data -> ServerSentEvent.builder()

.event("countDown") //和前端addEventListener监听的事件一一对应

.id(Long.toString(data.getT1())) //为每次发送设置一个id

.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 秒";

}

//报头设置为 "text/event-stream"，以便于发送事件流,这种写法等同于MediaType.TEXT_EVENT_STREAM_VALUE "text/event-stream;charset=UTF-8"

@GetMapping(value = "/retrieve",produces = MediaType.TEXT_EVENT_STREAM_VALUE)

@ResponseBody

public double retrieve() {

try {

//每0.5秒刷新数据

Thread.sleep(500);

} catch (InterruptedException e) {

e.printStackTrace();

}

//模拟股票实时变动数据

return Math.ceil(Math.random() * 10000);

}

最后，测试运行结果；

99023730017e

总结

虽然参考了很多资料，对于响应式编程还是很陌生，写个demo后，依旧没有感受到它的精华，基于WebFlux框架实现SSE事件，不难看出来还是基于长连接的，在实际场景中，基于长连接的推送事件是否适用，还值得再思考。

乐奇 Rokid 开放社区

更多推荐

从对话到行动：用 AIUI 构建「喝水节奏教练」

「喝水节奏教练」是专为Rokid AI眼镜设计的本地化饮水记录智能体，通过自然语言指令实现“记一杯”“查进度”“撤销错误”等操作，数据全程本地存储。基于意图驱动的单页面UI，结合语音识别与A2UI动态渲染，支持快速记录、实时核对与修正，不依赖云端或健康数据，专注轻量可信的饮水节奏管理。

乐奇 Rokid 开放社区

智能眼镜市场全景解析：从技术演进到产业格局

本文系统梳理智能眼镜市场，涵盖产品形态、技术演进、市场规模、产业链结构及竞争格局。指出智能眼镜正从音频交互向AR显示与空间计算演进，受AI大模型、供应链成熟与生态延伸驱动，呈现分层发展态势。虽面临场景刚需不足、续航重量矛盾与隐私挑战，但B端应用先行、AI原生设计与标准完善将推动行业迈向规模化渗透，未来需关注核心技术突破与真实使用数据。