Spring Boot가 SSE를 사용하여 프런트 엔드에 데이터를 푸시하는 방법-java지도 시간-php.cn

Spring Boot가 SSE를 사용하여 프런트 엔드에 데이터를 푸시하는 방법

WBOY

풀어 주다： 2023-05-10 17:31:06

앞으로

2966명이 탐색했습니다.

머리말

SSE는 단순히 서버가 적극적으로 데이터를 프런트엔드에 푸시하는 기술입니다. 즉, 프런트엔드가 서버에 데이터를 보낼 수 없다는 의미입니다. SSE는 메시지 푸시, 모니터링 및 서버 푸시 데이터만 필요한 기타 시나리오에 적합합니다. 다음은 Spring Boot를 사용하여 진행 데이터를 프런트 엔드에 푸시하는 간단한 시뮬레이션이며, 프런트 엔드 페이지에는 이를 수락한 후 진행률 표시줄이 표시됩니다.

Server side

Spring Boot에서 사용할 때 주의할 점은 Spring Web에서 제공하는 SseEmitter 클래스를 사용하여 동작시키는 것이 가장 좋습니다. 처음에는 인터넷에서 언급한 방법을 사용하여 Content를 설정했습니다. 텍스트 스트림에 입력하면 프런트 엔드가 매번 연결을 다시 생성하는 것으로 나타났습니다. 마지막으로 원하는 최종 효과를 얻으려면 다음 기사를 참조하십시오.

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 {

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

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

    public static SseEmitter connect(String userId){
        //设置超时时间，0表示不过期，默认是30秒，超过时间未完成会抛出异常
        SseEmitter sseEmitter = new SseEmitter(0L);
        //注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeOutCallBack(userId));
        sseEmitterMap.put(userId,sseEmitter);
        //数量+1
        count.getAndIncrement();
        log.info("create new sse connect ,current user:{}",userId);
        return sseEmitter;
    }
    /**
     * 给指定用户发消息
     */
    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();
            }
        }
    }

    /**
     * 想多人发送消息，组播
     */
    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);
            }
        });
    }
    /**
     * 群发消息
     */
    public static void batchSendMessage(String message, Set<String> userIds){
        userIds.forEach(userId->sendMessage(userId,message));
    }
    public static void removeUser(String userId){
        sseEmitterMap.remove(userId);
        //数量-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("结束连接,{}",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);
        };
    }
}

로그인 후 복사

위 클래스는 SSE 도구 클래스로 간주할 수 있습니다. 아래에서는

컨트롤러 레이어에서 생성합니다. 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));
        }
    }
}

로그인 후 복사

위의 연결은 sse에 연결하는 데 사용됩니다. 이때 연결이 생성되었으며, 다음 프로세스 인터페이스를 사용하여 데이터를 푸시합니다. 여기서는 진행률 표시줄의 효과를 얻기 위해 푸시되는 내용이 숫자입니다. 효과를 명확하게 하기 위해 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+&#39;%&#39;;
            document.getElementById("my-progress").value = event.data;
        }
    </script>
</head>
<body>
    <div id="result"></div>
    <progress  id="my-progress" value="0" max="100"></progress>
</body>
</html>

로그인 후 복사

입니다.

최종 효과:

Spring Boot가 SSE를 사용하여 프런트 엔드에 데이터를 푸시하는 방법