SSE 服务器发送事件详解

在 Http1.1 支持了长连接，请求头添加一个Connection: keep-alive即可

在这里我们借助异步请求来实现 sse 功能

因为后端可以不定时返回数据，所以我们需要注意的就是需要保持连接，不要返回一次数据之后就断开了；其次就是需要设置请求头Content-Type: text/event-stream;charset=UTF-8 （如果不是流的话会怎样？）

// 新建一个容器，保存连接，用于输出返回
private Map responseMap = new ConcurrentHashMap<>();

// 发送数据给客户端
private void writeData(String id, String msg, boolean over) throws IOException {
    PrintWriter writer = responseMap.get(id);
    if (writer == null) {
 return;
    }

    writer.println(msg);
    writer.flush();
    if (over) {
 responseMap.remove(id);
    }
}

// 推送
@ResponseBody
@GetMapping(path = "subscribe")
public WebAsyncTask subscribe(String id, HttpServletResponse response) {

    Callable callable = () -> {
 response.setHeader("Content-Type", "text/event-stream;charset=UTF-8");
 responseMap.put(id, response.getWriter());
 writeData(id, "订阅成功", false);
 while (true) {
     Thread.sleep(1000);
     if (!responseMap.containsKey(id)) {
  break;
     }
 }
 return null;
    };

    // 采用WebAsyncTask 返回 这样可以处理超时和错误 同时也可以指定使用的Excutor名称
    WebAsyncTask webAsyncTask = new WebAsyncTask<>(30000, callable);
    // 注意：onCompletion表示完成，不管你是否超时、是否抛出异常，这个函数都会执行的
    webAsyncTask.onCompletion(() -> System.out.println("程序[正常执行]完成的回调"));

    // 这两个返回的内容，最终都会放进response里面去===========
    webAsyncTask.onTimeout(() -> {
 responseMap.remove(id);
 System.out.println("超时了!!!");
 return null;
    });
    // 备注：这个是Spring5新增的
    webAsyncTask.onError(() -> {
 System.out.println("出现异常!!!");
 return null;
    });


    return webAsyncTask;
}

@ResponseBody
@GetMapping(path = "push")
public String pushData(String id, String content) throws IOException {
    writeData(id, content, false);
    return "over!";
}

@ResponseBody
@GetMapping(path = "over")
public String over(String id) throws IOException {
    writeData(id, "over", true);
    return "over!";
}

Content-Type: text/event-stream;charset=UTF-8
Cache-Control: no-cache
Connection: keep-alive

field:valuenn

@RestController
@RequestMapping(path = "sse")
public class SseRest {
    private static Map sseCache = new ConcurrentHashMap<>();
    @GetMapping(path = "subscribe")
    public SseEmitter push(String id) {
 // 超时时间设置为1小时
 SseEmitter sseEmitter = new SseEmitter(3600_000L);
 sseCache.put(id, sseEmitter);
 sseEmitter.onTimeout(() -> sseCache.remove(id));
 sseEmitter.onCompletion(() -> System.out.println("完成！！！"));
 return sseEmitter;
    }

    @GetMapping(path = "push")
    public String push(String id, String content) throws IOException {
 SseEmitter sseEmitter = sseCache.get(id);
 if (sseEmitter != null) {
     sseEmitter.send(content);
 }
 return "over";
    }

    @GetMapping(path = "over")
    public String over(String id) {
 SseEmitter sseEmitter = sseCache.get(id);
 if (sseEmitter != null) {
     sseEmitter.complete();
     sseCache.remove(id);
 }
 return "over";
    }
}

    


sse测试

@Controller
@RequestMapping(path = "sse")
public class SseRest {
    @GetMapping(path = "")
    public String index() {
 return "index.html";
    }

    @ResponseBody
    @GetMapping(path = "subscribe", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
    public SseEmitter push(String id) {
 // 超时时间设置为3s，用于演示客户端自动重连
 SseEmitter sseEmitter = new SseEmitter(1_000L);
 // 设置前端的重试时间为1s
 sseEmitter.send(SseEmitter.event().reconnectTime(1000).data("连接成功"));
 sseCache.put(id, sseEmitter);
 System.out.println("add " + id);
 sseEmitter.onTimeout(() -> {
     System.out.println(id + "超时");
     sseCache.remove(id);
 });
 sseEmitter.onCompletion(() -> System.out.println("完成！！！"));
 return sseEmitter;
    }
}