Technologieaustausch

Serverseitige Push-Datentechnologie SSE (Server-Send-Event).

2024-07-12

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

Serverseitige Push-Datentechnologie SSE (Server-Send-Event).

Sind Sie jemals auf eine Situation gestoßen, in der der Server aktiv Daten an den Client übertragen muss? Derzeit gibt es drei Lösungen.

  1. Der Client fragt nach aktualisierten Daten.
  2. Der Server und der Client stellen eine Socket-Verbindung für die bidirektionale Kommunikation her.
  3. Der Server stellt eine unidirektionale SSE-Verbindung mit dem Client her

Vergleich mehrerer Optionen:

  1. Umfrage:

    Der Client fordert durch häufige Anfragen Daten vom Server an, um einen ähnlichen Effekt wie Echtzeitaktualisierungen zu erzielen. Der Vorteil von Polling besteht darin, dass es einfach zu implementieren ist, jedoch den Server und das Netzwerk zusätzlich belastet und die Verzögerung hoch ist.

  2. WebSocket-Verbindung:

    Der Server stellt eine Socket-Verbindung mit dem Client zur Datenübertragung her. Die Socket-Übertragungsmethode ist Vollduplex. WebSocket ist eine lange Verbindung, die auf TCP basiert und eine leichte Datenübertragung mit geringer Latenz ermöglicht. Sie eignet sich sehr gut für Echtzeit-Kommunikationsszenarien und wird hauptsächlich für hochgradig interaktive Zwei-Wege-Kommunikation verwendet.

  3. SSE-Push:

    SSE (Server-Sent Events) ist eine Push-Technologie, die auf dem HTTP-Protokoll basiert und nur eine einseitige Kommunikation ermöglicht. Im Vergleich zu WebSocket ist SSE einfacher und leichter.

Im Folgenden finden Sie die Schritte und den Beispielcode für die Verwendung von SSE mit SpringBoot

  1. Konfigurationsabhängigkeiten

    	    <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
       <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    SSE wurde in Spring-Web integriert und kann daher direkt verwendet werden.

  2. Backend-Code

    import com.wry.wry_test.service.SseService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.validation.constraints.NotBlank;
import java.util.concurrent.CompletableFuture;

@RestController
@RequestMapping("/sse")
@Slf4j
@Validated
public class SseTestController {

    @Autowired
    private SseService service;


    @GetMapping("/testSse")
    public SseEmitter testSse(@RequestParam("clientId") @NotBlank(message = "客户端id不能为空") String clientId) {
        final SseEmitter emitter = service.getConn(clientId);
        CompletableFuture.runAsync(() -> {
            try {
                service.send(clientId);
                log.info("建立连接成功！clientId = {}", clientId);
            } catch (Exception e) {
                log.error("推送数据异常");
            }
        });
        return emitter;
    }


    @GetMapping("/sseConection")
    public SseEmitter createConnection(@RequestParam("clientId") @NotBlank(message = "客户端id不能为空") String clientId) {
        return service.getConn(clientId);
    }

    @GetMapping("/sendMsg")
    public void sendMsg(@RequestParam("clientId") String clientId) {
        try {
            // 异步发送消息
            CompletableFuture.runAsync(() -> {
                try {
                    service.send(clientId);
                } catch (Exception e) {
                    log.error("推送数据异常");
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/sendMsgToAll")
    public void sendMsgToAll() {
        try {
            //异步发送消息
            CompletableFuture.runAsync(() -> {
                try {
                    service.sendToAll();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @GetMapping("closeConn/{clientId}")
    public String closeConn(@PathVariable("clientId") @NotBlank(message = "客户端id不能为空") String clientId) {
        service.closeConn(clientId);
        return "连接已关闭";
    }


}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    package com.wry.wry_test.service;

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.validation.constraints.NotBlank;

public interface SseService {


    /**
     * 获取连接
     * @param clientId 客户端id
     * @return
     */
    SseEmitter getConn(String clientId);

    /**
     *  发送消息到指定客户端
     * @param clientId 客户端id
     * @throws Exception
     */
    void send(String clientId);

    /**
     * 发送消息到所有SSE客户端
     * @throws Exception
     */
    void sendToAll() throws Exception;

    /**
     * 关闭指定客户端的连接
     * @param clientId 客户端id
     */
    void closeConn(String clientId);
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    package com.wry.wry_test.service.impl;

import com.wry.wry_test.service.SseService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.validation.constraints.NotBlank;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


@Service
@Slf4j
public class SseServiceImpl implements SseService {

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

    @Override
    public SseEmitter getConn(@NotBlank String clientId) {
        final SseEmitter sseEmitter = SSE_CACHE.get(clientId);

        if (sseEmitter != null) {
            return sseEmitter;
        } else {
            // 设置连接超时时间，需要配合配置项 spring.mvc.async.request-timeout: 600000 一起使用
            final SseEmitter emitter = new SseEmitter(600_000L);
            // 注册超时回调，超时后触发
            emitter.onTimeout(() -> {
                log.info("连接已超时，正准备关闭，clientId = {}", clientId);
                SSE_CACHE.remove(clientId);
            });
            // 注册完成回调，调用 emitter.complete() 触发
            emitter.onCompletion(() -> {
                log.info("连接已关闭，正准备释放，clientId = {}", clientId);
                SSE_CACHE.remove(clientId);
                log.info("连接已释放，clientId = {}", clientId);
            });
            // 注册异常回调，调用 emitter.completeWithError() 触发
            emitter.onError(throwable -> {
                log.error("连接已异常，正准备关闭，clientId = {}", clientId, throwable);
                SSE_CACHE.remove(clientId);
            });

            SSE_CACHE.put(clientId, emitter);
            log.info("建立连接成功！clientId = {}", clientId);
            return emitter;
        }
    }

    /**
     * 模拟类似于 chatGPT 的流式推送回答
     *
     * @param clientId 客户端 id
     * @throws IOException 异常
     */
    @Override
    public void send(@NotBlank String clientId) {
        final SseEmitter emitter = SSE_CACHE.get(clientId);
        if (emitter == null) return;

        // 开始推送数据
        // todo 模拟推送数据
        for (int i = 0; i < 10000000; i++) {
            String msg = "SSE 测试数据";
            try {
                this.sseSend(emitter, msg, clientId);
                Thread.sleep(1000);
            } catch (Exception e) {
                log.error("推送数据异常", e);
                break;
            }
        }

        log.info("推送数据结束，clientId = {}", clientId);
        // 结束推流
        emitter.complete();
    }

    /**
     * 发送数据给所有连接
     */
    public void sendToAll() {
        List<SseEmitter> emitters = new ArrayList<>(SSE_CACHE.values());
        for (int i = 0; i < 10000000; i++) {
            String msg = "SSE 测试数据";
            this.sseSend(emitters, msg);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void closeConn(@NotBlank String clientId) {
        final SseEmitter sseEmitter = SSE_CACHE.get(clientId);
        if (sseEmitter != null) {
            sseEmitter.complete();
        }
    }

    /**
     * 推送数据封装
     *
     * @param emitter  sse长连接
     * @param data     发送数据
     * @param clientId 客户端id
     */
    private void sseSend(SseEmitter emitter, Object data, String clientId) {
        try {
            emitter.send(data);
            log.info("推送数据成功，clientId = {}", clientId);
        } catch (Exception e) {
            log.error("推送数据异常", e);
            throw new RuntimeException("推送数据异常");
        }
    }

    /**
     * 推送数据封装
     *
     * @param emitter sse长连接
     * @param data    发送数据
     */
    private void sseSend(List<SseEmitter> emitter, Object data) {

        emitter.forEach(e -> {
            try {
                e.send(data);
            } catch (IOException ioException) {
                log.error("推送数据异常", ioException);
            }
        });
        log.info("推送数据成功");
    }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142

    Der Implementierungseffekt ist wie folgt: Der Server überträgt kontinuierlich Daten an das Front-End, und das Front-End kann auch die Schnittstelle aufrufen, um die Verbindung aktiv zu schließen.

    Bild-20240710180401231

Anwendbare Szenarien: Da es sich bei SSE um eine unidirektionale Kommunikation auf der Serverseite handelt, eignet es sich für Verbindungen, die eine unidirektionale Persistenz erfordern. Zum Beispiel:

  • ChatGPT lädt Sitzungsdaten in Echtzeit
  • Datei-Download, Dateien asynchron über SSE herunterladen
  • Serverseitiger Daten-Push in Echtzeit

Persönliches Profil

Er widmet sich seit mehr als 30 Jahren der Technologieforschung und beherrscht verschiedene Sprachen wie Java, Linux, Javascript, PHP, CSS usw. Er hat viele Beiträge im Open-Source-Bereich geleistet Entwicklerdokumentationsstation, um einige Themen in der Technologieentwicklung als zukünftige Referenz zu teilen

meine Kontaktdaten

Post