<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

public class SseEmitterServer {
    private static final Logger LOGGER = LoggerFactory.getLogger(SseEmitterServer.class);
    /**
     * 當(dāng)前連接數(shù)
     */
    private static AtomicInteger count = new AtomicInteger(0);
    private static Map<Integer, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    public static SseEmitter connect(Integer userId){
        // 設(shè)置超時日期，0表示不過期
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注冊回調(diào)
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        sseEmitterMap.put(userId,sseEmitter);
        count.getAndIncrement();
        LOGGER.info("創(chuàng)建新SSE連接，連接用戶編號:{}",userId);
        LOGGER.info("現(xiàn)有連接用戶："+sseEmitterMap.keySet());
        return sseEmitter;
    }
    /**
     * 給指定用戶發(fā)信息
     */
    public static void sendMessage(Integer userId,String message){
        if (!sseEmitterMap.containsKey(userId)) {
            connect(userId);
        }
        try {
            sseEmitterMap.get(userId).send(message);
            LOGGER.info("給" + userId + "號發(fā)送消息：" + message);
        } catch (IOException e) {
            LOGGER.error("userId:{},發(fā)送信息出錯:{}", userId, e.getMessage());
            e.printStackTrace();
        }
    }
    /**
     * 群發(fā)消息
     */
    public static void batchSendMessage(String message){
        if (sseEmitterMap != null&&!sseEmitterMap.isEmpty()) {
            sseEmitterMap.forEach((k,v)->{
                try {
                    v.send(message, MediaType.APPLICATION_JSON);
                } catch (IOException e) {
                    LOGGER.error("userId:{},發(fā)送信息出錯:{}",k,e.getMessage());
                    e.printStackTrace();
                }
            });
        }
    }
    public static void batchSendMessage(Set<Integer> userIds,String message){
        userIds.forEach(userId->sendMessage(userId,message));
    }
    /**
     * 移出用戶
     */
    public static void removeUser(Integer userId){
        sseEmitterMap.remove(userId);
        count.getAndDecrement();
        LOGGER.info("remove user id:{}",userId);
        LOGGER.info("remain user id:"+sseEmitterMap.keySet());
    }
    public static List<Integer> getIds(){
        return new ArrayList<>(sseEmitterMap.keySet());
    }
    public static int getUserCount(){
        return count.intValue();
    }
    private static Runnable completionCallBack(Integer userId){
        return ()->{
          LOGGER.info("結(jié)束連接,{}",userId);
          removeUser(userId);
        };
    }
    private static Runnable timeoutCallBack(Integer userId){
        return ()->{
            LOGGER.info("連接超時,{}",userId);
            removeUser(userId);
        };
    }
    private static Consumer<Throwable> errorCallBack(Integer userId){
        return throwable -> {
            LOGGER.error("連接異常,{}",userId);
            removeUser(userId);
        };
    }
}

@RestController
@CrossOrigin(maxAge = 3600)
public class SseController {
    @RequestMapping(value = "/sse/connect/{id}",method = RequestMethod.GET)
    public SseEmitter connect(@PathVariable Integer id){
        SseEmitter sseEmitter = SseEmitterServer.connect(id);
        return sseEmitter;
    }
    /**
     * 向指定用戶發(fā)送消息
     */
    @RequestMapping(value = "/sse/send/{id}", method = RequestMethod.GET)
    public EiInfo sendMsg(@PathVariable Integer id,@RequestParam("message") String message) {
        EiInfo eiInfo = new EiInfo();
        SseEmitterServer.sendMessage(id,message);
        eiInfo.sysSetMsg("向"+id+"號用戶發(fā)送信息，"+message+"，消息發(fā)送成功");
        return eiInfo;
    }
    /**
     * 向所有用戶發(fā)送消息
     */
    @RequestMapping(value = "/sse/send/all", method = RequestMethod.GET)
    public EiInfo sendMsg2AllUser(@RequestParam("message") String message) {
        EiInfo eiInfo = new EiInfo();
        SseEmitterServer.batchSendMessage(message);
        eiInfo.sysSetMsg("向所有用戶發(fā)送信息，"+message+"，消息發(fā)送成功");
        return eiInfo;
    }
    /**
     * 關(guān)閉用戶連接
     */
    @RequestMapping(value = "/sse/close/{id}", method = RequestMethod.GET)
    public EiInfo closeSse(@PathVariable Integer id) {
        EiInfo eiInfo = new EiInfo();
        SseEmitterServer.removeUser(id);
        eiInfo.sysSetMsg("關(guān)閉"+id+"號連接。當(dāng)前連接用戶有："+SseEmitterServer.getIds());
        return eiInfo;
    }
}