鐵猛

java通過SSE實現消息推送

一、什么是SSE

嚴格地說，HTTP 協議無法做到服務器主動推送信息。但是，有一種變通方法，就是服務器向客戶端聲明，接下來要發送的是流信息（streaming）。

也就是說，發送的不是一次性的數據包，而是一個數據流，會連續不斷地發送過來。這時，客戶端不會關閉連接，會一直等著服務器發過來的新的數據流，視頻播放就是這樣的例子。本質上，這種通信就是以流信息的方式，完成一次用時很長的下載。

SSE 就是利用這種機制，使用流信息向瀏覽器推送信息。它基于 HTTP 協議，目前除了 IE，其他瀏覽器都支持。

二、如何在java中實現SSE

在Spring Boot項目中，無需額外引入特定的依賴，因為Spring Web MVC模塊已經內置了對SSE的支持。

1、編寫SSE服務，來進行創建鏈接和發送消息

package com.smart3dmap.prjpntmanage.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
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;

@Slf4j
@Service
public class SSEService {
    private static final Map<String,SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    public SseEmitter crateSse(String uid) {
        SseEmitter sseEmitter = new SseEmitter(0L);
        sseEmitter.onCompletion(() -> {
            log.info("[{}]結束鏈接" , uid);
            sseEmitterMap.remove(uid);
        });
        sseEmitter.onTimeout(() -> {
            log.info("[{}]鏈接超時",uid);
        });
        sseEmitter.onError(throwable -> {
            try{
                log.info("[{}]鏈接異常，{}",uid,throwable.toString());
                sseEmitter.send(SseEmitter.event()
                        .id(uid)
                        .name("發生異常")
                        .data("發生異常請重試")
                        .reconnectTime(3000));
                sseEmitterMap.put(uid,sseEmitter);
            }catch (IOException e){
                e.printStackTrace();
            }
        });
        try{
            sseEmitter.send(SseEmitter.event().reconnectTime(5000));
        }catch (IOException e){
            e.printStackTrace();
        }
        sseEmitterMap.put(uid,sseEmitter);
        log.info("[{}]創建sse連接成功!",uid);
        return sseEmitter;
    }

    public boolean sendMessage(String uid,String messageId,String message){
        if(StringUtils.isEmpty(message)){
            log.info("[{}]參數異常，msg為空",uid);
            return false;
        }
        SseEmitter sseEmitter = sseEmitterMap.get(uid);
        if(sseEmitter == null){
            log.info("[{}]sse連接不存在",uid);
            return  false;
        }
        try{
            sseEmitter.send(SseEmitter.event().id(messageId).reconnectTime(60000).data(message));
            log.info("用戶{},消息ID：{}，推送成功：{}",uid,messageId,message);
            return true;
        }catch (IOException e){
            sseEmitterMap.remove(uid);
            log.info("用戶{},消息ID：{}，消息推送失敗：{}",uid,messageId,message);
            sseEmitter.complete();
            return false;
        }
    }

    public void closeSse(String uid){
        if(sseEmitterMap.containsKey(uid)){
            SseEmitter sseEmitter = sseEmitterMap.get(uid);
            sseEmitter.complete();
            sseEmitterMap.remove(uid);
        }else {
            log.info("用戶{}連接已關閉",uid);
        }
    }

}

2、編寫對應的Controller來實現具體業務

package com.test.controller;

import cn.hutool.core.util.IdUtil;
import com.test.service.SSEService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@Controller
@RequestMapping("test")
public class TestController {
    @Autowired
    private SSEService sseService;
    @GetMapping("createSse")
    @CrossOrigin
    public SseEmitter createSse(String uid)
    {
        return sseService.crateSse(uid);
    }

    @GetMapping("sendMsg")
    @ResponseBody
    @CrossOrigin
    public String sseChat(String uid){
        for (int i = 0; i < 10; i++) {
            sseService.sendMessage(uid,"消息"+i,IdUtil.fastUUID().replace("-",""));
        }
        return "OK";
    }
    @GetMapping("closeSse")
    @CrossOrigin
    public void closeSse(String uid){
        sseService.closeSse(uid);
    }
}

三、前端實現消息的監聽

<!doctype html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>SSE消息推送監聽</title>
</head>
<body>
    <div id="conMsg"></div>
<script>
    let uid = 1;
    let chat = document.getElementById("conMsg");
    if(window.EventSource){
        var eventSource = new EventSource(`http://127.0.0.1:8090/test/createSse?uid=${uid}`);
        eventSource.onopen = ()=>{
            console.log("鏈接成功");
        }
        eventSource.onmessage = (ev)=>{
            if(ev.data){
                chat.innerHTML += ev.data+"<br>";
            }
        }
        eventSource.onerror = ()=>{
            console.log("sse鏈接失敗")
        }
    }else{
        alert("當前瀏覽器不支持sse")
    }
</script>
</body>
</html>

posted on 2024-11-04 15:25 鐵猛閱讀(3926) 評論(1) 收藏舉報

刷新頁面返回頂部

鐵猛

java通過SSE實現消息推送

導航

公告