快捷導(dǎo)航

SpringBoot使用SSE進(jìn)行實(shí)時通知前端的實(shí)現(xiàn)代碼

更新時間：2023年06月06日 10:17:41 作者：誰不想飛舞青春

這篇文章主要介紹了SpringBoot使用SSE進(jìn)行實(shí)時通知前端,本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下

說明

項(xiàng)目有個需求是要實(shí)時通知前端，告訴前端這個任務(wù)加載好了。然后想了2個方案，一種是用websocket進(jìn)行長連接，一種是使用SSE(Sever Send Event)，是HTTP協(xié)議中的一種，Content-Type為text/event-stream，能夠保持長連接。
websocket是前端既能向后端發(fā)送消息，后端也能向前端發(fā)送消息。
SSE是只能后端向前端發(fā)送消息。
因?yàn)橹恍枰蠖送ㄖ晕疫@里選擇了使用SSE實(shí)現(xiàn)。
這里先做個筆記，怕以后忘記怎么使用。

maven依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.3</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.project</groupId>
    <artifactId>test</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>test</name>
    <description>test</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <!--web依賴,內(nèi)嵌入tomcat,SSE依賴于該jar包,只要有該依賴就能使用SSE-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--lombok依賴,用來對象省略寫set、get方法-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.22</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

SSE工具類代碼

package com.etone.project.utils;
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;
@Slf4j
public class SseEmitterServer {
    /**
     * 當(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));
        //這個onError在springbooot低版本沒有這個方法,公司springboot1.4.2版本，沒有這個方法，可以進(jìn)行注釋。
        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);
        };
    }
}

Controller測試代碼

package com.project.test.controller;
import com.hjl.test.util.SseEmitterServer;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.HashMap;
import java.util.Map;
@RestController
@RequestMapping(value = "/test")
public class TestController {
    //sse連接接口
    @GetMapping (value = "/sse/connect/{id}")
    public SseEmitter connect(@PathVariable String id){
        return SseEmitterServer.connect(id);
    }
    //sse向指定用戶發(fā)送消息接口
    @GetMapping (value = "/sse/send/{id}")
    public Map<String,Object> send(@PathVariable String id,@RequestParam(value = "message", required = false) String message){
        Map<String,Object> returnMap = new HashMap<>();
        //向指定用戶發(fā)送信息
        SseEmitterServer.sendMessage(id,message);
        returnMap.put("message","向id為"+id+"的用戶發(fā)送："+message+"成功!");
        returnMap.put("status","200");
        returnMap.put("result",null);
        return returnMap;
    }
    //sse向所有已連接用戶發(fā)送消息接口
    @GetMapping (value = "/sse/batchSend")
    public Map<String,Object> batchSend(@RequestParam(value = "message", required = false) String message){
        Map<String,Object> returnMap = new HashMap<>();
        //向指定用戶發(fā)送信息
        SseEmitterServer.batchSendMessage(message);
        returnMap.put("message",message+"消息發(fā)送成功!");
        returnMap.put("status","200");
        returnMap.put("result",null);
        return returnMap;
    }
    //sse關(guān)閉接口
    @GetMapping (value = "/sse/close/{id}")
    public Map<String,Object> close(@PathVariable String id){
        Map<String,Object> returnMap = new HashMap<>();
        //移除id
        SseEmitterServer.removeUser(id);
        System.out.println("當(dāng)前連接用戶id:"+SseEmitterServer.getIds());
        returnMap.put("message","連接關(guān)閉成功!");
        returnMap.put("status","200");
        returnMap.put("result",null);
        return returnMap;
    }
}

測試結(jié)果如下:

這里測試SSE連接，就像正常接口那樣請求就行。
本地調(diào)用接口/sse/connect/1如下:
這里我連接2個用戶，用來模擬向指定用戶id發(fā)送信息和批量向已連接的用戶發(fā)送消。

后端服務(wù)打印如下:

本地調(diào)用接口/sse/send/1如下:

用戶1的結(jié)果如下，發(fā)現(xiàn)它收到了消息:

用戶2沒有收到結(jié)果，如下:

本地調(diào)用接口/sse/batchSend如下:
批量向所有已經(jīng)連接的用戶發(fā)送消息。

用戶1結(jié)果如下，發(fā)現(xiàn)接收到了消息:

用戶2結(jié)果如下，發(fā)現(xiàn)也接收到了消息:

測試結(jié)果都符合預(yù)期。
點(diǎn)擊postman的close按鈕，關(guān)閉連接：

發(fā)現(xiàn)前端連接雖然關(guān)閉了，但是后端實(shí)際還在連接中，根本沒有移除用戶的提示:

所以這里還需要自己手動寫關(guān)閉接口測試。
本地調(diào)用接口/sse/close/1如下:

可以看到把用戶id為1的給移除了，只剩用戶2還在連接中。

這里所有測試完成，結(jié)果符合預(yù)期。

注意

將超時時間由原來的0改為默認(rèn)的30秒，會報錯。

測試結(jié)果如下:

這里直接出現(xiàn)了一個異常:org.springframework.web.context.request.async.AsyncRequestTimeoutException
甚至連接都斷開了。

將springboot降為低版本如1.4.2.RELEASE。

使用postman進(jìn)行測試的時候，發(fā)現(xiàn)它不是一直在請求中：如下:

將Springboot降為1.4.2.RELEASE

springboot的1.4.2.RELEASE版本沒有onError方法，需要注釋掉。

postman測試如下:
低版本測試的時候發(fā)現(xiàn)它有一個這個連接可以直接看到，而使用springboot版本2.x版本就發(fā)現(xiàn)它一直處于發(fā)送請求的狀態(tài)，什么時候后端向前端發(fā)送了消息，它就顯示這個。
springboot的1.4.2.RELEASE版本結(jié)果：