package com.zillion.aggregate.app.controller;

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("[{}]結(jié)束鏈接" , 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("發(fā)生異常")
                        .data("發(fā)生異常請重試")
                        .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("[{}]創(chuàng)建sse連接成功!",uid);
        return sseEmitter;
    }

    public boolean sendMessage(String uid,String messageId,String message){
        if(StringUtils.isEmpty(message)){
            log.info("[{}]參數(shù)異常，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("用戶{}連接已關(guān)閉",uid);
        }
    }
}

package com.zillion.aggregate.app.controller;

import cn.hutool.core.util.IdUtil;
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;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Controller
@RequestMapping("/aggregate/api/pay")
public class TestController {

    private static final Map<String,Boolean> SEND_MAP = new ConcurrentHashMap<>();

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

    @GetMapping("/sendMsg")
    @ResponseBody
    @CrossOrigin
    public SseEmitter sendMsg(@RequestParam("uid") String uid) throws InterruptedException {
        SseEmitter sseEmitter = sseService.crateSse(uid);
        if (SEND_MAP.get(uid)==null ||  !SEND_MAP.get(uid)){
             new Thread(()->{
                 int i=0;
                 while (true){
                     try {
                         i++;
                         String message = "uid:"+uid+" number:"+i+" message:"+IdUtil.fastUUID().replace("-", "");
                         sseService.sendMessage(uid,"消息"+i,message);
                         SEND_MAP.put(uid,true);
                         Thread.sleep(1000);
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                         closeSse(uid);
                     }
                 }
             }).start();
         }

        return sseEmitter;
    }

    @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消息推送監(jiān)聽</title>
</head>
<body>
    <div id="conMsg"></div>
<script>
    let uid = 1;
    let chat = document.getElementById("conMsg");
    if(window.EventSource){
        var eventSource = new EventSource(`http://localhost:9001/aggregate/aggregate/api/pay/sendMsg?interfaceId=CEDB297CECCC9DCBAD348204ACDD5BAD&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>