WebFlux
12 分钟 读完 (大约 1869 个字)

服务端主动推送数据，除了 WebSocket 你还能想到啥？

[TOC]

在上篇文章中，松哥和大家分享了 WebFlux 的基本用法，小伙伴们已经了解到使用 WebFlux 我们的返回值可以是 Mono 也可以是 Flux，如果是 Flux，由于 Flux 中包含多个元素，所以我们需要设置响应的 Content-Type 为 text/event-stream。考虑到很多小伙伴还没用过 text/event-stream，所以今天松哥再撸一篇文章来和大家聊聊 text/event-stream。

1.SSE

首先我们来看一个概念叫做 SSE。

SSE 全称是 Server-Sent Events，它的作用和 WebSocket 的作用相似，都是建立浏览器与服务器之间的通信渠道，然后服务器向浏览器推送信息，不同的是，WebSocket 是一种全双工通信协议，而 SSE 则是一种单工通信协议，即使用 SSE 只能服务器向浏览器推送信息流，浏览器如果向服务器发送信息，就是一个普通的 HTTP 请求。

使用 SSE，当服务端给客户端响应的时候，他不是发送一个一次性数据包，而是会发送一个数据流，这个时候客户端的连接不会关闭，会一直等待服务端发送过来的数据流，我们常见的视频播放其实就是这样的例子。

SSE 和 WebSocket 主要有如下区别：

  • SSE 使用 HTTP 协议，现有的服务器软件都支持。WebSocket 是一个独立协议。
  • SSE 属于轻量级，使用简单；WebSocket 协议相对复杂。
  • SSE 默认支持断线重连，WebSocket 需要自己实现。
  • SSE 一般只用来传送文本，二进制数据需要编码后传送，WebSocket 默认支持传送二进制数据。
  • SSE 支持自定义发送的消息类型。

说了这么多，可能大家还是有点懵，接下来松哥通过一个简单的例子来向大家展示 SSE 的用法。

2.开发服务端

根据第一小节的描述，大家也能看出来，SSE 其实和框架没有关系，所以这里松哥就创建一个普通的 Java Web 项目，用最最基本的 Servlet 来向大家演示 SSE 的功能。

首先我们创建一个 SseServlet，内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@WebServlet(urlPatterns = "/sse")
public class SseServlet extends HttpServlet {
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        doPost(req,resp);
    }

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        resp.setContentType("text/event-stream;charset=utf-8");
        PrintWriter out = resp.getWriter();
        for (int i = 0; i < 10; i++) {
            out.write("data: 江南一点雨:" + i+"\n\n");
            out.flush();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

往出写 10 条数据，每写一条就睡眠 1 秒钟。代码并不难，但是这里有几个细节需要注意下：

  1. 响应的 Content-Type 记得设置为 text/event-stream，这是关键。
  2. 每一次发送的信息，由若干个 message 组成，每个 message 之间用 \n\n 分隔，每个 message 内部由若干行组成。在上面的案例中，每一个 for 循环中就是发送一个 message。
  3. 每一行的数据格式是 ：[field]: value\n。field 有四种不同取值：
    • data：data 用来表示数据内容，就像我们上面的例子。
    • id：id 相当于是每一条数据的唯一编号，浏览器用 lastEventId 属性读取这个值。一旦连接断线，浏览器会发送一个 HTTP 头，里面包含一个特殊的 Last-Event-ID 头信息，将这个值发送回来，用来帮助服务器端重建连接。因此，这个头信息可以被视为一种同步机制。
    • event：event 字段表示自定义的事件类型，默认是 message 事件。
    • retry：服务器可以用 retry 字段，指定浏览器重新发起连接的时间间隔。

开发完成后，我们启动服务端访问 /sse 接口来看看效果：

可以看到，客户端每隔 1 秒就能收到服务端的数据。

3.开发客户端

前面是一个服务端的案例，接下来我们来看看客户端的案例，新建一个 html 页面，添加如下 js：

1
2
3
4
5
6
7
8
9
10
11
var es = new EventSource("/sse");
es.onopen = function (e) {
    console.log("open")
};
es.onmessage = function (e) {
    console.log(e.data);
}
es.onerror = function (e) {
    console.log("error")
    es.close()
}

关于上面这段代码：

  1. 首先新建一个 EventSource 对象，参数就是服务端的地址。它还有一个可选的参数，可选参数重可以描述是否将 Cookie 一起发送出去 var es = new EventSource("/es", { withCredentials: true });（可在跨域时使用该参数）。
  2. 当建立连接后，就会触发 onopen 函数，当收到服务端发送来的消息，就会触发 onmessage 函数，当连接出错的时候，就会触发 onerror 函数。
  3. es.close 表示关闭 SSE 连接。

这三种类型的事件，我们还可以通过如下方式来定义：

1
2
3
4
5
6
7
8
9
10
11
var es = new EventSource("/sse");
es.addEventListener("open", function (e) {
    console.log("open");
})
es.addEventListener("message", function (e) {
    console.log(e.data);
})
es.addEventListener("error", function (e) {
    console.log("error")
    es.close();
})

效果与上面的一致，我们来看看运行效果图：

消息接收完后，会触发 onerror 事件，此时我们可以关闭 SSE 连接，否则就会从头开始继续接收数据。

4.自定义事件

我们也可以自定义 SSE 事件。

先来看服务端如何自定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@WebServlet(urlPatterns = "/sse")
public class SseServlet extends HttpServlet {
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        doPost(req, resp);
    }

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        resp.setContentType("text/event-stream;charset=utf-8");
        PrintWriter out = resp.getWriter();
        for (int i = 0; i < 10; i++) {
            out.write("event:javaboy\n");
            out.write("data: 江南一点雨:" + i + "\n\n");
            out.flush();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

如上，在每一行消息之前添加 out.write("event:javaboy\n"); 表示自定义事件类型，当然我们也可以添加事件 id，方式如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@WebServlet(urlPatterns = "/sse")
public class SseServlet extends HttpServlet {
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        doPost(req, resp);
    }

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        resp.setContentType("text/event-stream;charset=utf-8");
        PrintWriter out = resp.getWriter();
        for (int i = 0; i < 10; i++) {
            out.write("event:javaboy\n");
            out.write("id:" + i + "\n");
            out.write("data: 江南一点雨:" + i + "\n\n");
            out.flush();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

服务端定义完成后，接下来我们再来看看前端该如何接收数据：

1
2
3
4
5
6
7
8
9
10
11
var es = new EventSource("/sse");
es.addEventListener("open", function (e) {
    console.log("open");
})
es.addEventListener("javaboy", function (e) {
    console.log(e.data, e.lastEventId, e);
})
es.addEventListener("error", function (e) {
    console.log("error")
    es.close();
})

此时在 addEventListener 方法中，输入自定义的事件名称，然后在回调函数中处理事件。

可以通过 e.lastEventId 访问到消息的 id。

5.小结

好啦，今天主要通过几个简单的例子向大家展示 text/event-stream 以及 SSE 相关的知识点，相信大家在学完之后对 WebFlux 中返回值为 Flux 的接口会有更深的理解，读完本文，再去看昨天的文章【WebFlux 初体验】，应该会更香。

参考资料：http://www.ruanyifeng.com/blog/2017/05/server-sent_events.html

# WebFlux

喜欢这篇文章吗？扫码关注公众号【江南一点雨】，【江南一点雨】专注于 SPRING BOOT+微服务以及前后端分离技术，每天推送原创技术干货，关注后回复 JAVA，领取松哥为你精心准备的 JAVA 干货!

觉得不错，可以给小编加个鸡腿

用 WebFlux 写个 CURD 是什么体验？
WebFlux 初体验

评论

扫码关注微信公众号

分类

标签云

2019-article Actuator Blog Bug CAS CORS Chrome Docker Druid Ehcache ElasticJob ElasticSearch ElasticSerach Eureka FastDFS Flowable Flyway Freemarker Git GitHub HDC Https I18N IDEA IDEA，EasyCode IntelliJ IDEA JDK16 JWT Java JavaWeb JdbcTemplate Jedis Jenkins Jpa LiveReload Log Lombok MacBook MacBookPro Mail Maven MongoDB MyBatis MyBatisPlus MyCat MySQL MySQL8 Nacos Nginx OAuth2 PicGo Platform Plugin RabbitMQ Redis Resume SSM Shiro Spring Spring Boot Spring Cloud Spring Data Spring Security Spring6 SpringBoot SpringBoot3 SpringCloud SpringData SpringDataJpa SpringMVC SpringSecurity Swagger Swagger2 Swagger3 TLS Tencent Thymeleaf TienChin Transaction Video ViewResolver Vue Vue3 Web WebFlux blog docker domain elasticjob es gRPC hdz hexo huawei i18n idea java life mail nginx redis router saga seata tcc transaction vhr video vue vuepress weixin xa xxl-job yaml 七牛云 中间件 主从复制 事务 侃侃而谈 全局ID 分布式 分布式事务 分布式锁 前后端分离 华为 单体应用 外包 学习资源 定时任务 工具 干货汇总 延迟队列 微人事 微服务 杂谈 条件注解 死信队列 消息过期 消息驱动 源码解析 源码解读 生活 私活 程序人生 程序员 腾讯 自学 视频 视频教程 资料 逆向工程 面试

最新文章

归档

友链

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×