Springboot 集成 SSE 向前端推送消息

Sse推送

• • Sse介绍
  • 特点分析
  • 应用场景
  • Spring Boot 集成
  • 测试
  • 前端代码

Sse介绍

sse(Server Sent Event)，直译为服务器发送事件，顾名思义，也就是客户端可以获取到服务器发送的事件

我们常见的 http 交互方式是客户端发起请求，服务端响应，然后一次请求完毕；但是在 sse 的场景下，客户端发起请求，连接一直保持，服务端有数据就可以返回数据给客户端，这个返回可以是多次间隔的方式

特点分析

SSE 最大的特点，可以简单规划为两个

• 长连接
• 服务端可以向客户端推送信息

了解 websocket 的小伙伴，可能也知道它也是长连接，可以推送信息，但是它们有一个明显的区别

sse 是单通道，只能服务端向客户端发消息；而 webscoket 是双通道
那么为什么有了 webscoket 还要搞出一个 sse 呢？既然存在，必然有着它的优越之处

应用场景

从 sse 的特点出发，我们可以大致的判断出它的应用场景，需要轮询获取服务端最新数据的 case 下，多半是可以用它的

比如显示当前网站在线的实时人数，法币汇率显示当前实时汇率，电商大促的实时成交额等等…

Spring Boot 集成

项目结构

使用到的依赖

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-all</artifactId>
    <version>5.7.16</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
</dependency>

消息实体

package com.ddz.sse.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
 * 消息体
 *
 * @author Lenovo
 * @date 2022/5/6
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class MessageVo {
    /**
     * 客户端id
     */
    private String clientId;
    /**
     * 传输数据体(json)
     */
    private String data;
}

接口

package com.ddz.sse.service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
/**
 * @author Lenovo
 * @date 2022/5/6
 */
public interface SseEmitterService {
    /**
     * 创建连接
     *
     * @param clientId 客户端ID
     */
    SseEmitter createConnect(String clientId);
    /**
     * 根据客户端id获取SseEmitter对象
     *
     * @param clientId 客户端ID
     */
    SseEmitter getSseEmitterByClientId(String clientId);
    /**
     * 发送消息给所有客户端
     *
     * @param msg 消息内容
     */
    void sendMessageToAllClient(String msg);
    /**
     * 给指定客户端发送消息
     *
     * @param clientId 客户端ID
     * @param msg      消息内容
     */
    void sendMessageToOneClient(String clientId, String msg);
    /**
     * 关闭连接
     *
     * @param clientId 客户端ID
     */
    void closeConnect(String clientId);
}

实现类

package com.ddz.sse.service.impl;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpStatus;
import com.ddz.sse.entity.MessageVo;
import com.ddz.sse.service.SseEmitterService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
@Slf4j
@Service
public class SseEmitterServiceImpl implements SseEmitterService {
    /**
     * 容器，保存连接，用于输出返回 ;可使用其他方法实现
     */
    private static final Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
    /**
     * 根据客户端id获取SseEmitter对象
     *
     * @param clientId 客户端ID
     */
    @Override
    public SseEmitter getSseEmitterByClientId(String clientId) {
        return sseCache.get(clientId);
    }
    /**
     * 创建连接
     *
     * @param clientId 客户端ID
     */
    @Override
    public SseEmitter createConnect(String clientId) {
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 是否需要给客户端推送ID
        if (StrUtil.isBlank(clientId)) {
            clientId = IdUtil.simpleUUID();
        }
        // 注册回调
        sseEmitter.onCompletion(completionCallBack(clientId));     // 长链接完成后回调接口(即关闭连接时调用)
        sseEmitter.onTimeout(timeoutCallBack(clientId));        // 连接超时回调
        sseEmitter.onError(errorCallBack(clientId));          // 推送消息异常时，回调方法
        sseCache.put(clientId, sseEmitter);
        log.info("创建新的sse连接，当前用户：{}    累计用户:{}", clientId, sseCache.size());
        try {
            // 注册成功返回用户信息
            sseEmitter.send(SseEmitter.event().id(String.valueOf(HttpStatus.HTTP_CREATED)).data(clientId, MediaType.APPLICATION_JSON));
        } catch (IOException e) {
            log.error("创建长链接异常，客户端ID:{}   异常信息:{}", clientId, e.getMessage());
        }
        return sseEmitter;
    }
    /**
     * 发送消息给所有客户端
     *
     * @param msg 消息内容
     */
    @Override
    public void sendMessageToAllClient(String msg) {
        if (MapUtil.isEmpty(sseCache)) {
            return;
        }
        // 判断发送的消息是否为空
        for (Map.Entry<String, SseEmitter> entry : sseCache.entrySet()) {
            MessageVo messageVo = new MessageVo();
            messageVo.setClientId(entry.getKey());
            messageVo.setData(msg);
            sendMsgToClientByClientId(entry.getKey(), messageVo, entry.getValue());
        }
    }
    /**
     * 给指定客户端发送消息
     *
     * @param clientId 客户端ID
     * @param msg      消息内容
     */
    @Override
    public void sendMessageToOneClient(String clientId, String msg) {
        MessageVo messageVo = new MessageVo(clientId, msg);
        sendMsgToClientByClientId(clientId, messageVo, sseCache.get(clientId));
    }
    /**
     * 关闭连接
     *
     * @param clientId 客户端ID
     */
    @Override
    public void closeConnect(String clientId) {
        SseEmitter sseEmitter = sseCache.get(clientId);
        if (sseEmitter != null) {
            sseEmitter.complete();
            removeUser(clientId);
        }
    }
    /**
     * 推送消息到客户端
     * 此处做了推送失败后，重试推送机制，可根据自己业务进行修改
     *
     * @param clientId  客户端ID
     * @param messageVo 推送信息，此处结合具体业务，定义自己的返回值即可
     **/
    private void sendMsgToClientByClientId(String clientId, MessageVo messageVo, SseEmitter sseEmitter) {
        if (sseEmitter == null) {
            log.error("推送消息失败：客户端{}未创建长链接,失败消息:{}",
                    clientId, messageVo.toString());
            return;
        }
        SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(String.valueOf(HttpStatus.HTTP_OK))
                .data(messageVo, MediaType.APPLICATION_JSON);
        try {
            sseEmitter.send(sendData);
        } catch (IOException e) {
            // 推送消息失败，记录错误日志，进行重推
            log.error("推送消息失败：{},尝试进行重推", messageVo.toString());
            boolean isSuccess = true;
            // 推送消息失败后，每隔10s推送一次，推送5次
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(10000);
                    sseEmitter = sseCache.get(clientId);
                    if (sseEmitter == null) {
                        log.error("{}的第{}次消息重推失败，未创建长链接", clientId, i + 1);
                        continue;
                    }
                    sseEmitter.send(sendData);
                } catch (Exception ex) {
                    log.error("{}的第{}次消息重推失败", clientId, i + 1, ex);
                    continue;
                }
                log.info("{}的第{}次消息重推成功,{}", clientId, i + 1, messageVo.toString());
                return;
            }
        }
    }
    /**
     * 长链接完成后回调接口(即关闭连接时调用)
     *
     * @param clientId 客户端ID
     **/
    private Runnable completionCallBack(String clientId) {
        return () -> {
            log.info("结束连接：{}", clientId);
            removeUser(clientId);
        };
    }
    /**
     * 连接超时时调用
     *
     * @param clientId 客户端ID
     **/
    private Runnable timeoutCallBack(String clientId) {
        return () -> {
            log.info("连接超时：{}", clientId);
            removeUser(clientId);
        };
    }
    /**
     * 推送消息异常时，回调方法
     *
     * @param clientId 客户端ID
     **/
    private Consumer<Throwable> errorCallBack(String clientId) {
        return throwable -> {
            log.error("SseEmitterServiceImpl[errorCallBack]：连接异常,客户端ID:{}", clientId);
            // 推送消息失败后，每隔10s推送一次，推送5次
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(10000);
                    SseEmitter sseEmitter = sseCache.get(clientId);
                    if (sseEmitter == null) {
                        log.error("SseEmitterServiceImpl[errorCallBack]：第{}次消息重推失败,未获取到 {} 对应的长链接", i + 1, clientId);
                        continue;
                    }
                    sseEmitter.send("失败后重新推送");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }
    /**
     * 移除用户连接
     *
     * @param clientId 客户端ID
     **/
    private void removeUser(String clientId) {
        sseCache.remove(clientId);
        log.info("SseEmitterServiceImpl[removeUser]:移除用户：{}", clientId);
    }
}

控制器类

package com.ddz.sse.controller;
import com.ddz.sse.entity.MessageVo;
import com.ddz.sse.service.SseEmitterService;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import javax.annotation.Resource;
/**
 * SSE长链接
 */
@RestController
@RequestMapping("/sse")
public class SseEmitterController {
    @Resource
    private SseEmitterService sseEmitterService;
    @CrossOrigin
    @GetMapping("/createConnect")
    public SseEmitter createConnect(String clientId) {
        return sseEmitterService.createConnect(clientId);
    }
    @CrossOrigin
    @PostMapping("/broadcast")
    public void sendMessageToAllClient(@RequestBody(required = false) String msg) {
        sseEmitterService.sendMessageToAllClient(msg);
    }
    @CrossOrigin
    @PostMapping("/sendMessage")
    public void sendMessageToOneClient(@RequestBody(required = false) MessageVo messageVo) {
        if (messageVo.getClientId().isEmpty()) {
            return;
        }
        sseEmitterService.sendMessageToOneClient(messageVo.getClientId(), messageVo.getData());
    }
    @CrossOrigin
    @GetMapping("/closeConnect")
    public void closeConnect(@RequestParam(required = true) String clientId) {
        sseEmitterService.closeConnect(clientId);
    }
}

测试

自己创建clientid

使用系统创建clientid

这里浏览器一直在转圈说明是连接成功.
我们使用APIPost接口测试工具测试

这里ddz123的是能收到我们推送的消息的.

前端代码

<!doctype html>
<html lang="en">
<head>
    <title>Sse测试文档</title>
</head>
<body>
<div>sse测试</div>
<div id="data"></div>
</body>
</html>
<script>
    var source = new EventSource('http://192.168.1.128:8889/sse/createConnect?clientId=ddz');
    source.onmessage = function (event) {
        text = document.getElementById('data').innerText;
        text += '\n' + event.data;
        document.getElementById('data').innerText = text;
    };
    <!-- 添加一个开启回调 -->
    source.onopen = function (event) {
        text = document.getElementById('data').innerText;
        text += '\n 开启: ';
        console.log(event);
        document.getElementById('data').innerText = text;
    };
</script>

如果需要携带token请求,那就需要引入一个插件

npm install event-source-polyfill

我这里的是vue的代码;代码不完善只供参考一下

<template>
    <div>sse测试</div>
    <div id="data"></div>
</template>
<script>
import {
    EventSourcePolyfill
} from 'event-source-polyfill';
export default {
    name: 'sse',
    data() {
        return {
            token:'获取到的令牌'
        }
    },
    mounted() {
       },
    created() {
        this.createContent('http://192.168.1.128:8080/syscon/sse/createConnect','?clientId=ddz',this.token)
    },
    methods: {
        //创建连接
        createContent(url, params, token) {
            return new EventSourcePolyfill(url, params, {
                headers: {
                    Authorization: token
                }
            })
        },
    },
}
</script>

文章参考于https://www.cnblogs.com/yihuihui/p/12622729.html