HTTP 是客户端-服务器计算模型中的请求-响应协议。要开始交换,客户端向服务器提交请求。为了完成交换,服务器向客户端返回响应。服务器只能向一个客户端发送响应 (发出请求的那个) 。在 HTTP 协议中,客户端是消息交换的发起者。
有些场景需要由服务端主动推送消息给客户端。实现这一点的方法之一是允许服务器在发布/订阅计算模型中向客户端推送消息。要开始交换,客户端从服务器订阅消息。在交换期间,服务器向许多订阅的客户端发送消息(一旦它们可用)。
服务器发送事件 (SSE) 是一种简单的技术,用于为特定的 Web 应用程序实现服务器到客户端的异步通信。
有多种技术允许客户端从服务器接收有关异步更新的消息。它们可以分为两类:客户端拉取和服务器推送。
在客户端拉取技术中,客户端会定期向服务器请求更新。服务器可以使用更新或尚未更新的特殊响应进行响应。有两种类型的客户端拉取:短轮询和长轮询。
客户端定期向服务器发送请求。如果服务器有更新,它会向客户端发送响应并关闭连接。如果服务器没有更新,它也会向客户端发送一个响应并关闭连接。
客户端向服务器发送请求。如果服务器有更新,它会向客户端发送响应并关闭连接。如果服务器没有更新,它会保持连接直到更新可用。当更新可用时,服务器向客户端发送响应并关闭连接。如果更新在某个超时时间内不可用,服务器会向客户端发送响应并关闭连接。
在服务器推送技术中,服务器在消息可用后立即主动向客户端发送消息。其中,有两种类型的服务器推送:SSE和 WebSocket。
SSE 是一种在基于浏览器的 Web 应用程序中仅从服务器向客户端发送文本消息的技术。SSE基于 HTTP 协议中的持久连接, 具有由 W3C 标准化的网络协议和 EventSource 客户端接口,作为 HTML5 标准套件的一部分。
WebSocket 是一种在 Web 应用程序中实现同时、双向、实时通信的技术。WebSocket 基于 HTTP 以外的协议(TCP),因此可能需要额外设置网络基础设施(代理服务器、NAT、防火墙等)。
客户端通过Http协议请求,在握手阶段升级为WebSocket协议。
要订阅服务器事件,客户端发出 GET 请求带有指定的header:
GET /sse HTTP/1.1
Accept: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
服务器应该使用带有标题的响应来确认订阅:
HTTP/1.1 200
Content-Type: text/event-stream;charset=UTF-8
Transfer-Encoding: chunked
订阅后,服务端在消息可用时立即发送给客户端。事件是采用 UTF-8 编码的文本消息。事件之间由两个换行符分隔\n\n。每个事件由一个或多个名称:值字段组成,由单个换行符\n 分隔。
在数据字段中,服务器可以发送事件数据
data: The first event.
data: The second event.
服务器可以发送唯一的事件标识符(id字段)。如果连接中断,客户端会自动重新连接并发送最后接收到的带有header的 Last-Event-ID 的事件 ID。
在事件字段中,服务器可以发送事件类型。服务器可以在同一个订阅中发送不同类型的事件,也可以不发送任何类型的事件。
event: type1
data: An event of type1.
event: type2
data: An event of type2.
data: An event without any type.
在重试字段中,服务器可以发送超时(以毫秒为单位),之后客户端应在连接中断时自动重新连接。如果未指定此字段,则标准应为 3000 毫秒。
retry: 1000
如果一行以冒号字符 : 开头,客户端应该忽略它。这可用于从服务器发送评论或防止某些代理服务器因超时关闭连接。
: ping
要打开连接,应创建一个 EventSource 对象。
var eventSource = new EventSource('/sse);
尽管 SSE 旨在将事件从服务器发送到客户端,但可以使用 GET 查询参数将数据从客户端传递到服务器。
var eventSource = new EventSource('/sse?event=type1);
...
eventSource.close();
eventSource = new EventSource('/sse?event=type1&event=type2);
...
要关闭连接,应调用方法 close()。
eventSource.close();
有表示连接状态的 readyState 属性:
要处理连接的建立,它应该订阅 onopen 事件处理程序。
eventSource.onopen = function () {
console.log('connection is established');
};
为了处理连接状态的一些异常或致命错误,它应该订阅 onerrror 事件处理程序。
eventSource.onerror = function (event) {
console.log('connection state: ' + eventSource.readyState + ', error: ' + event);
};
客户端接收消息并处理他们,可以使用onmessage方法
eventSource.onmessage = function (event) {
console.log('id: ' + event.lastEventId + ', data: ' + event.data);
};
SSE可被大多数浏览器支持:
Spring Web MVC 框架 5.2.0 是基于 Servlet 3.1 API 且用线程池实现异步应用程序. 所以应用能够被使用在 Servlet 3.1+ 的容器,比如:Tomcat 8.5 和 Jetty 9.3.
使用Spring MVC来发送事件:
示例:
@RestController
public class SseWebMvcController
private SseEmitter emitter;
@GetMapping(path="/sse", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
SseEmitter createConnection() {
emitter = new SseEmitter();
return emitter;
}
// in another thread
void sendEvents() {
try {
emitter.send("Alpha");
emitter.send("Omega");
emitter.complete();
} catch(Exception e) {
emitter.completeWithError(e);
}
}
}
在这个例子中,服务器每秒发送一个持续时间短的周期性事件流 - 一个有限的词流,直到词完成。
示例:
@Controller
@RequestMapping("/sse/mvc")
public class WordsController {
private static final String[] WORDS = "The quick brown fox jumps over the lazy dog.".split(" ");
private final ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
@GetMapping(path = "/words", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
SseEmitter getWords() {
SseEmitter emitter = new SseEmitter();
cachedThreadPool.execute(() -> {
try {
for (int i = 0; i < WORDS.length; i++) {
emitter.send(WORDS[i]);
TimeUnit.SECONDS.sleep(1);
}
emitter.complete();
} catch (Exception e) {
emitter.completeWithError(e);
}
});
return emitter;
}
}
运行效果:
客户端示例(words.html):
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Server-Sent Events client example with EventSource</title>
</head>
<body>
<script>
if (window.EventSource == null) {
alert('The browser does not support Server-Sent Events');
} else {
var eventSource = new EventSource('/sse/mvc/words');
eventSource.onopen = function () {
console.log('connection is established');
};
eventSource.onerror = function (error) {
console.log('connection state: ' + eventSource.readyState + ', error: ' + event);
};
eventSource.onmessage = function (event) {
console.log('id: ' + event.lastEventId + ', data: ' + event.data);
if (event.data.endsWith('.')) {
eventSource.close();
console.log('connection is closed');
}
};
}
</script>
</body>
</html>
运行效果:
在此示例中,服务器发送持久的周期性事件流 - 每秒可能无限的服务器性能信息流:
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@RestController
@RequestMapping("/sse/mvc")
public class LongEventController {
private final ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);
private SseEmitter emitter;
@PostConstruct
public void init() {
scheduledThreadPool.scheduleAtFixedRate(() -> {
try {
if (emitter != null) {
emitter.send(UUID.randomUUID().toString());
}
} catch (IOException e) {
e.printStackTrace();
}
}, 0, 1, TimeUnit.SECONDS);
}
@GetMapping(path = "/getEvents", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter getEvents() {
emitter = new SseEmitter();
return emitter;
}
}
效果预览(每秒输出一次):
非周期性是指没有固定的时间周期,可能由其他因素在任意时刻都可能触发,下面示例通过spring event来模拟触发因子。
@RestController
@RequestMapping("/sse/mvc")
public class EventController {
private SseEmitter emitter;
@Autowired
private ApplicationContext applicationContext;
/**
* 订阅事件通道
* @return
*/
@GetMapping(path = "/event", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter event() {
emitter = new SseEmitter();
return emitter;
}
/**
* 模拟某一事件触发动作
* @param eventType
*/
@GetMapping(path = "/trigger")
public void trigger(String eventType) {
applicationContext.publishEvent(new MyEvent(eventType));
}
/**
* 监听动作,发送给客户端数据
*/
@EventListener(classes = MyEvent.class)
public void triggerEvent(MyEvent event) throws IOException {
emitter.send(event);
}
}
效果:
模拟触发动作:调用 http://localhost:8080/sse/mvc/trigger?eventType=customer
客户端收到数据:
Spring Web Flux 框架 5.2.0 是基于 Reactive Streams API 且使用 event-loop 计算模型来实现异步java应用程序。此类应用程序可以在非阻塞 Web 服务器(例如 Netty 4.1 和 Undertow 1.4)和 Servlet 3.1+ 容器(例如 Tomcat 8.5 和 Jetty 9.3)上运行。
使用 Spring Web Flux 框架实现发送事件:
简单示例:
@RestController
public class ExampleController
@GetMapping(path="/sse", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> createConnectionAndSendEvents() {
return Flux.just("Alpha", "Omega");
}
}
和上面spring mvc的示例一样,也是每秒输出数据,实现如下:
@GetMapping(path = "/words", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> getWords() {
return Flux
.zip(Flux.just(WORDS), Flux.interval(Duration.ofSeconds(1)))
.map(Tuple2::getT1);
}
效果:
对比spring mvc的实现,我们改为flux实现,如下:
@GetMapping(path = "/getEvents", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> getEvents() {
return Flux
.interval(Duration.ofSeconds(1))
.map(sequence -> UUID.randomUUID().toString());
}
效果和上面是一样的,可以看出,reactive api是非常的简洁。
探索ChatGPT的使用过程中,我们发现GPT采用了流式数据返回的方式。理论上,这种情况可以通过全双工通信协议实现持久化连接,或者依赖于基于EventStream的事件流。然而,ChatGPT选择了后者,也就是本文即将深入探讨的SSE(Server-Sent Events)技术。
要理解这个选择,我们需要关注ChatGPT的使用场景。作为一个基于深度学习的大型语言模型,ChatGPT需要处理大量的自然语言数据,这无疑需要大量的计算资源和时间。相较于普通的读取数据库操作,其响应速度自然会慢许多。
对于这种可能需要长时间等待响应的对话场景,ChatGPT采用了一种巧妙的策略:它会将已经计算出的数据“推送”给用户,并利用SSE技术在计算过程中持续返回数据。这样做可以避免用户因等待时间过长而选择关闭页面。
SSE(Server-Sent Events)是一种Web技术,它允许服务器实时向客户端推送数据。相比于传统的轮询和长轮询机制,SSE提供了一种更高效且实时的数据推送方式。这种技术主要应用于构建实时应用,例如实时消息推送、股票行情更新等。
SSE是HTML5规范中的一个通信相关API,它主要包含两个部分:服务端与浏览器端的通信协议(基于HTTP协议),以及浏览器端JavaScript可使用的EventSource对象。
SSE运行在HTTP协议之上,它允许服务器以事件流(Event Stream)的形式将数据发送给客户端。客户端通过建立持久化的HTTP连接,并监听这个事件流,从而可以实时接收到服务器推送的数据。
SSE具有以下几个主要特点:
WebSocket是一种Web技术,用于实现实时双向通信,它与SSE(Server-Sent Events)在某些方面存在差异。以下是对两者的比较:
选择使用SSE还是WebSocket主要取决于具体的业务需求和场景。如果你只需要实现从服务器向客户端的单向数据推送,并且希望保持操作简便且兼容性好,那么SSE是一个理想的选择。然而,如果你需要实现双向通信,或者需要更高级的功能和控制,那么WebSocket可能会更适合你的需求。
以下是SSE(Server-Sent Events)的实现原理:
总结起来,SSE使用了基于文本和HTTP协议的简单机制,使得服务器能够实时地将数据推送到客户端,而无需客户端频繁地发起新的请求。
以下是在使用SSE(Server-Sent Events)技术进行实时数据推送时需要注意的几个关键点:
以上这些注意事项可以根据具体应用需求进行调整和优化。在实际应用中,确保服务器的稳定性、安全性和性能是非常重要的。同时,在处理SSE连接时,可以考虑适当的限流和安全控制措施,以防止滥用和恶意连接的出现。总的来说,使用SSE技术时需要全面考虑各个方面的因素,才能实现高效、稳定、安全的实时数据推送服务。
假设正在开发一个实时股票价格监控应用,需要将股票价格实时推送给客户端。以下为Spring Boot中集成SSE技术实现的场景示例代码。
首先,定义一个控制器来处理SSE请求和发送实时股票价格:
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.Random;
@RestController
public class StockController {
@GetMapping(value = "/stock-price", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamStockPrice() {
SseEmitter emitter = new SseEmitter();
// 模拟生成实时股票价格并推送给客户端
Random random = new Random();
new Thread(() -> {
try {
while (true) {
// 生成随机的股票价格
double price = 100 + random.nextDouble() * 10;
// 构造股票价格的消息
String message = String.format("%.2f", price);
// 发送消息给客户端
emitter.send(SseEmitter.event().data(message));
// 休眠1秒钟
Thread.sleep(1000);
}
} catch (Exception e) {
emitter.completeWithError(e);
}
}).start();
return emitter;
}
}
在上述代码中,定义了一个streamStockPrice()方法,该方法使用@GetMapping注解将/stock-price路径映射到该方法上,并指定produces = MediaType.TEXT_EVENT_STREAM_VALUE以表明该方法将产生SSE事件流。
在方法内部创建了一个SseEmitter对象作为事件发射器,并在一个单独的线程中不断生成随机的股票价格,并将价格转换为字符串形式发送给客户端。
通过emitter.send()方法发送的数据会被封装为SSE事件流的形式,客户端可以通过监听该事件流来实时接收股票价格。
在前端页面中,创建一个简单的HTML页面来展示实时股票价格:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>实时股票价格监控</title>
</head>
<body>
<h1>实时股票价格</h1>
<div id="stock-price"></div>
<script>
const eventSource = new EventSource('/stock-price');
eventSource.onmessage = function (event) {
document.getElementById('stock-price').innerHTML = event.data;
};
</script>
</body>
</html>
上述代码中,通过new EventSource('/stock-price')创建了一个EventSource对象,它与/stock-price路径建立SSE连接。然后,通过eventSource.onmessage定义了接收消息的回调函数,在收到新消息时更新页面上的股票价格。
通过以上代码,可以在浏览器中打开该HTML页面,它会建立与服务器的SSE连接,并实时接收并展示股票价格。这只是使用SSE实现实时数据推送的一个简单示例。在实践中,可以根据具体的业务需求和场景,进行更复杂和丰富的实现。
SSE(Server-Sent Events)是一种基于HTTP协议的轻量级实时通信技术,具备服务端推送、断线重连和简单轻量等优点。然而,它也存在一些限制,例如无法进行双向通信、连接数受限以及仅支持GET请求等。
在Web应用程序中,SSE可以实现各种即时数据推送功能,如股票在线数据更新、日志推送、实时显示聊天室人数等。
然而,需要注意的是,并非所有的实时推送场景都适合使用SSE。在需要处理高并发、高吞吐量和低延迟的场景下,WebSocket可能是更好的选择。而对于那些需要轻量级推送解决方案的场景,SSE可能会更加适合。
因此,在选择实时更新方案时,我们需要根据具体的需求和应用场景来做出决策。只有这样,我们才能确保选择的技术能够最大程度地满足我们的需求。
是 码农小胖哥。天天有编程干货分享。觉得写的不错。点个赞,转发一下,关注一下。本文为个人原创文章,转载请注明出处,非法转载抄袭将追究其责任。
场景
今天项目经理交给我一个开发任务。如果有人在前台下了订单就给后台仓库管理一个发货通知。也就是服务端触发一个事件,推送消息到客户端。
如果我用websocket来做还要搞个websocket服务器,而且还 有不少配置。websocket是全双工通信,单向通信简直是杀鸡用牛刀。用轮询吧,浪费服务器资源不说,还不一定实时,订单处理慢了岂不是怠慢了客户。有没有别的选择呢?当然有!
1.SSE推送技术
SSE全称Server-sent Events,是HTML 5 规范的一个组成部分,具体去MDN网站查看相关文档。该规范十分简单,主要由两个部分组成:第一个部分是服务器端与浏览器端之间的通讯协议,第二部分是在浏览器端可供 JavaScript 使用的 EventSource 对象。通讯协议是基于纯文本的简单协议。服务器响应的内容类型是“text/event-stream”。响应文本的内容可以看成是一个事件流,由不同的事件所组成。每个事件由类型和数据两部分组成,同时每个事件可以有一个可选的标识符。不同事件的内容之间通过仅包含回车符和换行符的空行(“\r\n”)来分隔。每个事件的数据可能由多行组成。
如上图所示,每个事件之间通过空行来分隔。每一行都是由键值对组成。如果键为空则表示该行为注释,会在处理时被忽略。例如第10行。
第1行表示一个只包含数据的事件。会按照默认事件走(message事件)。第3-4行代表一个附带eventID的事件。第6-8行代表一个自定义事件。第10-14行代表一个多行数据事件,多行数据由换行符链接
key定义有以下几种:
SSE只适用于高级浏览器,但是注意IE不直接支持。IE上的XMLHttpRequest对象不支持获取部分的响应内容,所以不支持。每次总有IE,怪不得快被淘汰了。
2. SSE VS Websocket
3. Spring Mvc中的SSE
Spring Mvc对SSE进行了支持。如果你要声明一个SSE连接。只需要在你的控制器声明一个如下接口:
必须必须返回SseEmitter对象,SseEmitter对象是Session级别的,如果你要点对点针对每个session要独立存储。如果你是广播可以共用一个SseEmitter对象。按照SSE规范也必须声明produces为"text/event-stream"。当你调用该接口的时候将建立起SSE连接。
你可以在另一个线程中调用SseEmitter的send方法向客户端发送事件。你也可以在发送事件后调用complete方法来关闭SSE连接。
4.浏览器端的EventSource
由于SSE 是HTML5规范。所以对于APP端必须有HTML才能支持。并且IE如果要支持需要使用一些兼容开发包,比如polyfill库。客户端因为只接受事件所以开发比较简单:
5.总结
今天介绍了SSE 服务端推送。和长轮训、comet、websocket相比而言比较轻量级。在一些需要服务器实时推送规模不大的业务场景实现更简单点。相信看了本文后你会很快入门。在实际开发中要根据业务对这几种推送进行技术选型。没有最好的只有最适合的。SSE对大多数开发者来说不够熟悉。相关代码码云仓库:
https://gitee.com/felord/sse-push
*请认真填写需求信息,我们会在24小时内与您取得联系。