WebSocket и SseEmitter для отправки сообщений SpringBoot

Spring Boot

учиться никто конец территория , и июнь общий нежелание .

использовать

  • Получайте последние данные с сервера в режиме реального времени
  • Просмотр хода выполнения и состояния выполнения запланированных задач
  • Восприятие пользователя: после изменения данных соответствующие пользователи получают информацию
  • Усилить пользовательский опыт: трудоемкий бизнес асинхронная обработка (импорт Excel и экспорт, сложные расчеты)

Внешний опрос

Этот метод прост в реализации, фронтенд проходитsetIntervalРегулярно запрашивайте интерфейс для получения последних данных.Этот метод можно использовать, когда требования к реальному времени не высоки, а частота обновления низкая. Но когда режим реального времени высок, нашЗапрос будет очень частым, потребление сервера очень большое, и данные на сервере могут не меняться каждый раз при выполнении запроса, в результате чего многие запросы становятся бессмысленными.

    setInterval(function () {
            // 请求接口操作
            // 。。。
        },
        3000
    );

webSocket

WebSocket основан наTCP-протоколДа, это полнодуплексная связь, сервер может отправлять информацию клиенту, а клиент также может отправлять инструкции серверу, что часто используется в чат-приложениях.

pom.xml

SpringBoot предоставляет стартер для веб-сокета

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>

класс конфигурации

инъекцияServerEndpointExporter, этот bean-компонент будет автоматически зарегистрирован и использован@ServerEndpointКонечная точка Websocket, объявленная аннотацией

@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

Сервер класс

Создайте класс обслуживания:

  • плюс@ServerEndpointАннотация, задайте служебный адрес точки подключения WebSocket.
  • СоздайтеAtomicIntegerИспользуется для записи количества подключений
  • СоздайтеConcurrentHashMapИспользуется для хранения информации соединения
  • @OnOpenВ аннотации указано, что этот метод вызывается после установления соединения
  • @OnCloseВ аннотации указано, что этот метод вызывается после отключения
  • @OnErrorВ аннотации указано, что этот метод вызывается при исключении соединения.
  • @OnMessageВ аннотации указано, что этот метод вызывается после получения клиентского сообщения.
  • Как создавать push-сообщения
  • Создайте метод для удаления соединения
@ServerEndpoint("/websocket/{userId}")
@Component
public class WebSocketServer {

    private final static Logger logger = LoggerFactory.getLogger(WebSocketServer.class);

    /**
     * 当前连接数
     */
    private static AtomicInteger count = new AtomicInteger(0);

    /**
     * 使用map对象，便于根据userId来获取对应的WebSocket，或者放redis里面
     */
    private static Map<String, WebSocketServer> websocketMap = new ConcurrentHashMap<>();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 对应的用户ID
     */
    private String userId = "";

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        try {
            this.session = session;
            this.userId = userId;
            websocketMap.put(userId, this);
            // 数量+1
            count.getAndIncrement();
            logger.info("websocket 新连接：{}", userId);
        } catch (Exception e) {
            logger.error("websocket 新建连接 IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        // 删除
        websocketMap.remove(this.userId);
        // 数量-1
        count.getAndDecrement();
        logger.info("close websocket : {}", this.userId);
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message) {
        logger.info("来自客户端{}的消息:{}", this.userId, message);
    }

    @OnError
    public void onError(Throwable error) {
        logger.info("websocket 发生错误,移除当前websocket:{},err:{}", this.userId, error.getMessage());
        websocketMap.remove(this.userId);
        // 数量-1
        count.getAndDecrement();
    }

    /**
     * 发送消息 (异步发送)
     *
     * @param message 消息主题
     */
    private void sendMessage(String message) {
        this.session.getAsyncRemote().sendText(message);
    }

    /**
     * 向指定用户发送信息
     *
     * @param userId 用户id
     * @param wsInfo 信息
     */
    public static void sendInfo(String userId, String wsInfo) {
        if (websocketMap.containsKey(userId)) {
            websocketMap.get(userId).sendMessage(wsInfo);
        }
    }

    /**
     * 群发消息
     */
    public static void batchSendInfo(String wsInfo, List<String> ids) {
        ids.forEach(userId -> sendInfo(userId, wsInfo));
    }

    /**
     * 群发所有人
     */
    public static void batchSendInfo(String wsInfo) {
        websocketMap.forEach((k, v) -> v.sendMessage(wsInfo));
    }

    /**
     * 获取当前连接信息
     */
    public static List<String> getIds() {
        return new ArrayList<>(websocketMap.keySet());
    }

    /**
     * 获取当前连接数量
     */
    public static int getUserCount() {
        return count.intValue();
    }
}

тестовый интерфейс

@RestController
@RequestMapping("/ws")
public class WebSocketController {

    @GetMapping("/push/{message}")
    public ResponseEntity<String> push(@PathVariable(name = "message") String message) {
        WebSocketServer.batchSendInfo(message);
        return ResponseEntity.ok("WebSocket 推送消息给所有人");
    }

}

html

существуетresources/staticСоздать подws.html, задайте адрес WebSocket для класса обслуживания@ServerEndpointАннотировать настроенный адрес

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>WebSocket</title>
</head>

<body>
<div id="message"></div>
</body>

<script>
    let websocket = null;

    // 用时间戳模拟登录用户
    const username = new Date().getTime();
    // alert(username)
    //判断当前浏览器是否支持WebSocket
    if ('WebSocket' in window) {
        console.log("浏览器支持Websocket");
        websocket = new WebSocket('ws://localhost:8080/websocket/' + username);
    } else {
        alert('当前浏览器 不支持 websocket');
    }

    //连接发生错误的回调方法
    websocket.onerror = function () {
        setMessageInnerHTML("WebSocket连接发生错误");
    };

    //连接成功建立的回调方法
    websocket.onopen = function () {
        setMessageInnerHTML("WebSocket连接成功");
    };

    //接收到消息的回调方法
    websocket.onmessage = function (event) {
        setMessageInnerHTML(event.data);
    };

    //连接关闭的回调方法
    websocket.onclose = function () {
        setMessageInnerHTML("WebSocket连接关闭");
    };

    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function () {
        closeWebSocket();
    };

    //关闭WebSocket连接
    function closeWebSocket() {
        websocket.close();
    }

    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }
</script>
</html>

контрольная работа

Запустите проект, посетитеhttp://localhost:8080/ws.htmlчтобы открыть соединение. Вызов интерфейса отправки сообщенийhttp://localhost:8080/ws/push/helloдля просмотра информации, отображаемой на веб-странице.

SseEmitter

SseEmitter — это технология, предоставляемая SpringMVC (4.2+), основанная наHTTP-протоколДа, по сравнению с WebSocket он более легковесный, но его можно отправить только с сервера на клиентоднонаправленныйОтправить сообщение. В SpringBoot мы можем использовать его без ссылки на другие банки.

Создать класс обслуживания

  • СоздайтеAtomicIntegerИспользуется для записи количества подключений
  • СоздайтеConcurrentHashMapИспользуется для хранения информации о подключении
  • Установить соединение: создать и вернутьSseEmitterк переднему концу. Установите тайм-аут на 0, чтобы он никогда не истекал
  • Установите метод обратного вызова для завершения соединенияcompletionCallBack
  • Метод обратного вызова для настройки тайм-аута соединенияtimeoutCallBack
  • Установить метод обратного вызова для исключения подключенияerrorCallBack
  • Как создавать push-сообщенияSseEmitter.send()
  • Создайте метод для удаления соединения
public class SseEmitterServer {

    private static final Logger logger = LoggerFactory.getLogger(SseEmitterServer.class);

    /**
     * 当前连接数
     */
    private static AtomicInteger count = new AtomicInteger(0);

    /**
     * 使用map对象，便于根据userId来获取对应的SseEmitter，或者放redis里面
     */
    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    /**
     * 创建用户连接并返回 SseEmitter
     *
     * @param userId 用户ID
     * @return SseEmitter
     */
    public static SseEmitter connect(String userId) {
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        sseEmitterMap.put(userId, sseEmitter);
        // 数量+1
        count.getAndIncrement();
        logger.info("创建新的sse连接，当前用户：{}", userId);
        return sseEmitter;
    }

    /**
     * 给指定用户发送信息
     */
    public static void sendMessage(String userId, String message) {
        if (sseEmitterMap.containsKey(userId)) {
            try {
                // sseEmitterMap.get(userId).send(message, MediaType.APPLICATION_JSON);
                sseEmitterMap.get(userId).send(message);
            } catch (IOException e) {
                logger.error("用户[{}]推送异常:{}", userId, e.getMessage());
                removeUser(userId);
            }
        }
    }

    /**
     * 群发消息
     */
    public static void batchSendMessage(String wsInfo, List<String> ids) {
        ids.forEach(userId -> sendMessage(wsInfo, userId));
    }

    /**
     * 群发所有人
     */
    public static void batchSendMessage(String wsInfo) {
        sseEmitterMap.forEach((k, v) -> {
            try {
                v.send(wsInfo, MediaType.APPLICATION_JSON);
            } catch (IOException e) {
                logger.error("用户[{}]推送异常:{}", k, e.getMessage());
                removeUser(k);
            }
        });
    }

    /**
     * 移除用户连接
     */
    public static void removeUser(String userId) {
        sseEmitterMap.remove(userId);
        // 数量-1
        count.getAndDecrement();
        logger.info("移除用户：{}", userId);
    }

    /**
     * 获取当前连接信息
     */
    public static List<String> getIds() {
        return new ArrayList<>(sseEmitterMap.keySet());
    }

    /**
     * 获取当前连接数量
     */
    public static int getUserCount() {
        return count.intValue();
    }

    private static Runnable completionCallBack(String userId) {
        return () -> {
            logger.info("结束连接：{}", userId);
            removeUser(userId);
        };
    }

    private static Runnable timeoutCallBack(String userId) {
        return () -> {
            logger.info("连接超时：{}", userId);
            removeUser(userId);
        };
    }

    private static Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            logger.info("连接异常：{}", userId);
            removeUser(userId);
        };
    }

}

тестовый интерфейс

@RestController
@RequestMapping("/sse")
public class SseEmitterController {
    /**
     * 用于创建连接
     */
    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId) {
        return SseEmitterServer.connect(userId);
    }

    @GetMapping("/push/{message}")
    public ResponseEntity<String> push(@PathVariable(name = "message") String message) {
        SseEmitterServer.batchSendMessage(message);
        return ResponseEntity.ok("WebSocket 推送消息给所有人");
    }

}

html

существуетresources/staticСоздать подws.html, установите адрес EventSource на адрес, по которому было создано соединение

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SseEmitter</title>
</head>
<body>
<button onclick="closeSse()">关闭连接</button>
<div id="message"></div>
</body>
<script>
    let source = null;

    // 用时间戳模拟登录用户
    const userId = new Date().getTime();

    if (!!window.EventSource) {

        // 建立连接
        source = new EventSource('http://localhost:8080/sse/connect/' + userId);

        /**
         * 连接一旦建立，就会触发open事件
         * 另一种写法：source.onopen = function (event) {}
         */
        source.addEventListener('open', function (e) {
            setMessageInnerHTML("建立连接。。。");
        }, false);

        /**
         * 客户端收到服务器发来的数据
         * 另一种写法：source.onmessage = function (event) {}
         */
        source.addEventListener('message', function (e) {
            setMessageInnerHTML(e.data);
        });


        /**
         * 如果发生通信错误（比如连接中断），就会触发error事件
         * 或者：
         * 另一种写法：source.onerror = function (event) {}
         */
        source.addEventListener('error', function (e) {
            if (e.readyState === EventSource.CLOSED) {
                setMessageInnerHTML("连接关闭");
            } else {
                console.log(e);
            }
        }, false);

    } else {
        setMessageInnerHTML("你的浏览器不支持SSE");
    }

    // 监听窗口关闭事件，主动去关闭sse连接，如果服务端设置永不过期，浏览器关闭后手动清理服务端数据
    window.onbeforeunload = function () {
        closeSse();
    };

    // 关闭Sse连接
    function closeSse() {
        source.close();
        const httpRequest = new XMLHttpRequest();
        httpRequest.open('GET', 'http://localhost:8080/sse/close/' + userId, true);
        httpRequest.send();
        console.log("close");
    }

    // 将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }
</script>
</html>

контрольная работа

Запустите проект, посетите веб-страницуhttp://localhost:8080/sse.htmlустановить соединение. Вызов интерфейса отправки сообщенияhttp://localhost:8080/sse/push/hello, чтобы просмотреть информацию, отображаемую на веб-странице.

Доступ к исходному коду

Весь код загружен на Github для быстрого доступа

>>>>>> WebSocket и SseEmitter для отправки сообщений

ежедневные комплименты

Это не легко создать, если вы найдете это полезным,попросить лайкслужба поддержки

Поиск внимания

Категории