快捷導(dǎo)航

Spring?Boot?使用?SSE?方式向前端推送數(shù)據(jù)詳解

更新時間：2022年08月10日 16:09:06 作者：@胡海龍

這篇文章主要介紹了Spring?Boot?使用SSE方式向前端推送數(shù)據(jù)詳解，SSE簡單的來說就是服務(wù)器主動向前端推送數(shù)據(jù)的一種技術(shù)，它是單向的，也就是說前端是不能向服務(wù)器發(fā)送數(shù)據(jù)的

前言

SSE簡單的來說就是服務(wù)器主動向前端推送數(shù)據(jù)的一種技術(shù)，它是單向的，也就是說前端是不能向服務(wù)器發(fā)送數(shù)據(jù)的。SSE適用于消息推送，監(jiān)控等只需要服務(wù)器推送數(shù)據(jù)的場景中，下面是使用Spring Boot 來實現(xiàn)一個簡單的模擬向前端推動進(jìn)度數(shù)據(jù)，前端頁面接受后展示進(jìn)度條。

服務(wù)端

在Spring Boot中使用時需要注意，最好使用Spring Web 提供的SseEmitter這個類來進(jìn)行操作，我在剛開始時使用網(wǎng)上說的將Content-Type設(shè)置為text-stream這種方式發(fā)現(xiàn)每次前端每次都會重新創(chuàng)建接。最后參考該文實現(xiàn)了最終想要的效果：

SSE工具類

SSEServer.java

package vip.huhailong.catchat.sse;

import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @author Huhailong
 */
@Slf4j
public class SSEServer {

    /**
     * 當(dāng)前連接數(shù)
     */
    private static AtomicInteger count = new AtomicInteger(0);

    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    public static SseEmitter connect(String userId){
        //設(shè)置超時時間，0表示不過期，默認(rèn)是30秒，超過時間未完成會拋出異常
        SseEmitter sseEmitter = new SseEmitter(0L);
        //注冊回調(diào)
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeOutCallBack(userId));
        sseEmitterMap.put(userId,sseEmitter);
        //數(shù)量+1
        count.getAndIncrement();
        log.info("create new sse connect ,current user:{}",userId);
        return sseEmitter;
    }
    /**
     * 給指定用戶發(fā)消息
     */
    public static void sendMessage(String userId, String message){
        if(sseEmitterMap.containsKey(userId)){
            try{
                sseEmitterMap.get(userId).send(message);
            }catch (IOException e){
                log.error("user id:{}, send message error:{}",userId,e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * 想多人發(fā)送消息，組播
     */
    public static void groupSendMessage(String groupId, String message){
        if(sseEmitterMap!=null&&!sseEmitterMap.isEmpty()){
            sseEmitterMap.forEach((k,v) -> {
                try{
                    if(k.startsWith(groupId)){
                        v.send(message, MediaType.APPLICATION_JSON);
                    }
                }catch (IOException e){
                    log.error("user id:{}, send message error:{}",groupId,message);
                    removeUser(k);
                }
            });
        }
    }
    public static void batchSendMessage(String message) {
        sseEmitterMap.forEach((k,v)->{
            try{
                v.send(message,MediaType.APPLICATION_JSON);
            }catch (IOException e){
                log.error("user id:{}, send message error:{}",k,e.getMessage());
                removeUser(k);
            }
        });
    }
    /**
     * 群發(fā)消息
     */
    public static void batchSendMessage(String message, Set<String> userIds){
        userIds.forEach(userId->sendMessage(userId,message));
    }
    public static void removeUser(String userId){
        sseEmitterMap.remove(userId);
        //數(shù)量-1
        count.getAndDecrement();
        log.info("remove user id:{}",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 () -> {
            log.info("結(jié)束連接,{}",userId);
            removeUser(userId);
        };
    }
    private static Runnable timeOutCallBack(String userId){
        return ()->{
            log.info("連接超時,{}",userId);
            removeUser(userId);
        };
    }
    private static Consumer<Throwable> errorCallBack(String userId){
        return throwable -> {
            log.error("連接異常，{}",userId);
            removeUser(userId);
        };
    }
}

上面這個類可以把它當(dāng)作一個SSE的工具類，下面我們使用一下它

在Controller層創(chuàng)建 SSEController.java

package vip.huhailong.catchat.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import vip.huhailong.catchat.sse.SSEServer;

/**
 * @author Huhailong
 */
@Slf4j
@RestController
@CrossOrigin
@RequestMapping("/sse")
public class SSEController {

    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId){
        return SSEServer.connect(userId);
    }

    @GetMapping("/process")
    public void sendMessage() throws InterruptedException {
        for(int i=0; i<=100; i++){
            if(i>50&&i<70){
                Thread.sleep(500L);
            }else{
                Thread.sleep(100L);
            }
            SSEServer.batchSendMessage(String.valueOf(i));
        }
    }
}

上面的connect是用來連接sse的，它返回一個SseEmitter實例，這時候連接就已經(jīng)創(chuàng)建了，然后下面的process接口是用來推送數(shù)據(jù)的，我這里是準(zhǔn)備讓前端實現(xiàn)一個進(jìn)度條的效果，所以推送的是數(shù)字，為了效果明顯，我在推送到50到70的時候速度放慢，其余都是100ms

前端代碼

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Home</title>
    <script>
        let data = new EventSource("/cat-chat/sse/connect/huhailong")
        data.onmessage = function(event){
            console.log("test=>",event)
            document.getElementById("result").innerText = event.data+'%';
            document.getElementById("my-progress").value = event.data;
        }
    </script>
</head>
<body>
    <div id="result"></div>
    <progress style="width: 300px" id="my-progress" value="0" max="100"></progress>
</body>
</html>

最終效果：