● 心跳检测机制

心跳检测机制就是客户端在与服务器连接时，定时判断客户端是否活着，防止僵尸连接、假死连接，如果客户端挂了但服务器以为他还活着那就会造成资源浪费，只有客户端关闭发送-1给后端的nio的buffer读取到才是正常关闭不会造成资源浪费

心跳检测就是定时客户端或服务器端向另一方发送心跳包，如果服务端定时检测到某一连接长期不发心跳包而又有长连接维护，那么就直接关闭该连接，因为已经判定为死亡，心跳包并不能证明客户端还活着，只能证明连接还在，如果连接不在了那么心跳包也不会发了，自然就判定死亡了

• 误判现象：
  但如果客户端阻塞了，定期没有发送心跳包，服务端认定客户端没有连接了，直接杀死，那么等客户端不再阻塞就发现自己被误杀了
• 假死现象：
  如果客户端已经死亡了，崩溃了，但是底层的定时器线程还活着，服务端就会判定他还活着，系统就会持续维持这个连接
• 僵尸连接：
  也别称为TCP半开连接，比如用户的wifi断开了，服务器还以为他还活着，这时候服务器可以对这个连接写入信息，但是不会得到ACK，或者客户端已经崩溃了，OS没来得及通知服务器，因为tcp连接超时要等很久，默认两个小时
• 心跳检测机制检查的是连接健康而非业务健康
• 心跳检测机制核心目的就是检测目标连接的健康性，当客户端连接不可达时，如果不做心跳那么服务器系统的OS还会一直记录着客户端的tcp连接信息，这会导致fd、内存等很多资源的浪费，他就是一个快速版的tcp超时判断

○ 系统内核维护长连接细节与原理

名词解释：

• TCP：客户端与服务器之间通信的协议，客户端服务器必须按照这个协议来通信，比如四挥三握，数据确保对方要收到不丢不乱等
• SYN（Synchronize同步）：申请建立连接，发起连接，申请建立连接
• ACK（Acknowledge确认）：表示对方收到了，然后给你的回执说你过关！就可以开始说话了
• FIN（Finish）：客户端说完了，要关闭了，正常关闭才会发送FIN，比如-1
• DF（File Descriptor文件描述符）：代表OS内核中的某一个成员，比如一个tcp连接，一个buffer，一个管道等

当客户端浏览器发送一次数据到后端服务器时，这时候后端通过网卡tcp连接发送数据到后端，后端主从reactor读取到数据并注册连接，这时候后端不会去维护这个客户端的长连接，它只是负责记录连接信息状态与生命周期，而真正维护长联机的是客户端与服务器的系统内核

流程：

• 浏览器前端发送一次数据（SYN），此时就已经在TCP协议内的规则里了，到服务器
• 三次握手：
  第一次：客户端给服务器发SYN说我要建立连接
  第二次：服务器收到了SYN并通知ACK前端说我知道了，批准建立连接
  第三次：客户端通知服务器ACK，我知道了，可以实时通信了
• 这时候客户端的OS（Operating System）和服务器的OS都保存了这个TCP连接，这就是建立长连接的过程，客户端发数据给服务器首先这个数据会被网卡发送到OS记录的TCP连接的buffer内
• 服务器的nio拿数据只是打开TCP的buffer从这里面拿，他的read事件也只是再问OS，buffer里有没有东西我可不可以拿而已，他并不去维护这个连接，他拿数据只是通过FD打开OS内核的某一个成员可能是某一个tcp连接的buffer
• 当客户端正常关闭就会给服务器发送一个FIN，此时FIN的值为-1，服务器的socketChannel.read(buffer)收到之后就会判定为客户端已关闭

● WebSocket

WebSocket是客户端对服务器的全双工通信工具，当客户端通过WebSocket连接服务器时建立一个长连接来维持通信，他可以做心跳检测，群聊功能等
他在前端vue和后端spring都可以用，前端使用WebSocket会在请求头中标注Upgrade: websocket，而它在后端中是一个多实例注解，每一个连接对应一个WebSocket实例，这个WebSocket实例被tomcat的WebSocket容器管理，如果是Spring的WebSocket则是全局单例

对比项	Spring WebSocket	@ServerEndpoint
谁创建实例	Spring	Tomcat
实例模型	单例	每连接一个
Spring 依赖注入	✅ 原生支持	❌ 需特殊处理
底层	Tomcat	Tomcat

也就是说一个连接进来之后为这个连接建立session，然后WebSocket并不维护长连接而是os系统内核维护，这个session和tcp连接绑定用来记录连接信息，当一个tcp连接有数据后，WebSocket就会通过session知到是谁发的来的数据，这样保持的长连接，他并不是被spring直接调用，而是被tomcat的nio模型回调使用因为不管是@ServerEndPoint还是spring的WebSocket都是注册到tomcat的WebSocket容器内

流程：

• 前端浏览器发送一次WebSocket请求
• 前端浏览器的OS会记录这次的TCP连接
• 数据通过网卡发送到目标ip的服务器OS，此时服务器OS会记录这次的TCP连接
• 服务器tomcat的read事件检测到OS内的某个tcp连接的buffer内有数据，就会channel.read读取
• 读取到之后进行处理把这个请求封装为java的request对象，在此时tomcat会检查这个请求体内的Upgrade: websocket发现是WebSocket连接，就不调用servlet.service了，直接把这个请求体给spring的HandlerMapping
• 他会把这个请求分发给spring容器的WebSocket的Bean让他处理而不是交给DisPatcherServlet的controller处理了

使用：
建立客户端与服务器的双向通信，使用服务器ping\客户端pong来进行心跳检测机制，这样会避免假死连接，因为客户端假死后他的底层定时器线程会继续工作，这时候也会发送pong，但是不能接受ping，如果服务器发送ping客户端不接说明假死了，直接判定死亡

• 客户端轮询发 ping：客户端主动给服务器发一个 “极简的检测包”（不是业务数据）
• 服务器会检测session集合，这个session就是客户端连接的代表，如果这个连接已经很久没有发心跳包了就代表他死了，给他直接杀死使用session.close()这个方法会通告tomcat的nio关闭channel连接并且删除OS内核的tcp
• 注：如果使用Spring Security需要在Security的AuthenticationEntryPoint拦截配置中把/wx设为白名单因为tomcat把WebSocket请求丢给spring的HandlerMapping后依然会走tomcat的过滤器和spring的拦截器，所以/wx需要被放行，他也会经过jwtAuthticationTokenFilter的，但是它并不能拿到Security的上下文线程对象，因为WebSocket和用户请求不在同一个线程中，他是独立线程的

导入依赖：

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

创建配置类：
配置类配置了首次与客户端握手的WebSocket请求拦截器，这个拦截器主要作用就是看看客户端带没带着token来，还配置了WebSocket的实现类

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Resource
    private HeartWebSocket heartWebSocket;

    @Resource
    private HandShakeInterceptor handShakeInterceptor;

    /**
     * 注册WebSocket
     */
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry
                .addHandler(heartWebSocket, "/wx") //添加WebSocket与url的映射关系 相当于RequestMapping了
                .addInterceptors(handShakeInterceptor) //添加握手拦截器
                .setAllowedOriginPatterns("*"); // 开发环境用*，生产环境填前端域名（如http://localhost:5173）
    }
}

拦截器具体实现：

@Component
public class HandShakeInterceptor implements HandshakeInterceptor {

    @Resource
    private JwtProperties jwtProperties;

    /**
     * 拦截之前
     */
    @Override
    public boolean beforeHandshake(ServerHttpRequest request,
                                   ServerHttpResponse response,
                                   WebSocketHandler wsHandler,
                                   Map<String, Object> attributes) throws Exception {

        String token = null;
        // 获取请求参数：ws://localhost:8080/ws?token=xxx
        String query = request.getURI().getQuery();
        if (query != null && query.contains("token=")) {
            token = query.split("token=")[1];
        }

        return token != null && JwtUtil.validateToken(jwtProperties.getSecretKey(), token);
    }

    /**
     * 握手后
     */
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {

    }
}

○ DelayQueue

DelayQueue延迟队列，他是判断客户端是否已经死亡的核心，它的核心作用就是存放一个带有延迟时间的队列，只有当队列内的元素超时后，才会被take()取出，他是一个接口

public interface Delayed extends Comparable<Delayed> {
    // 1. 返回当前元素的剩余延迟时间（单位：纳秒）
    long getDelay(TimeUnit unit);
    
    // 2. 比较方法（用于优先级排序，通常按延迟时间排序）
    int compareTo(Delayed o);
}

getDelay()：告诉队列 “这个元素还需要等多久才能被取出”，返回值≤0 时表示延迟到期；
compareTo()：队列根据这个方法排序，确保延迟最短的元素在队首。

也就是说放入的expireTime - 当前时间 如果是负数说明到期了 就该被取出了 当然 这个expireTime通常是被放入时的系统时间+超时时间，因为在做倒计时计算

创建一个实现Delayed接口的实现类模型pojo：

@AllArgsConstructor
@NoArgsConstructor
@Data
public class WebSocketTimeOut implements Delayed {

    private Long sessionId;

    private Long expireTime;

    private Long version;

    @Override
    public long getDelay(@NotNull TimeUnit unit) {
        return unit.convert(expireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(@NotNull Delayed o) {
        return Long.compare(this.getDelay(TimeUnit.MILLISECONDS),
                o.getDelay(TimeUnit.MILLISECONDS));
    }
}

DelayQueue队列的实现：
流程：

• 当一个新连接到来时直接建立心跳包register(session)
• 放入sessionMap与versions并且DelayQueue.offer()
• WebSocketSessionManage被创建后会调用初始化方法PostConstruct的initTimeOutCheck这时候会创建一个守护线程，他跟随JVM进程
• 这个线程主要是阻塞遍历每一个Queue的task，进行超时检验，如果超时了就会从queue中提取出来，因为这是延迟队列，它内部自动判断倒计时
• 当客户端每次发送心跳就会重复执行上述方法
• 直到客户端非正常结束，这时候就会判定为僵尸连接，被handlerTimeOutTask杀死

versions的作用（重点）：

• 这个version干的活，因为这个queue.offer每一次提交的是一个新的独立的task，也就是每一次前端发一个心跳包 他就是一个新的task，那一个连接就可能发很多个心跳包，如果没有这个version当第一个task超时的时候 他一定会超时因为没人去更新他的时间，然后他就会执行session.close()这里非常可怕 他就会直接把一个还活着的连接给关掉，所以这时候version就发挥作用了，每一个task存储的都是不一样的version当前面的version还是1后面的就变成了2，那么再去versions内拿指定sessionId的version就一定会不相等 就说明这个是旧任务直接丢弃即可，如果这个连接非正常关闭了，并且也不心跳了，那么session就会被删掉version也不会再更新了，这时候v.get就==task.getVersion()了 这时候就代表这是一个真正的假死用户 直接关闭连接即可
• 重要误点：每一次offer都是新的WebSocketTimeOut对象而不是一个对象持续进行时间的更新
• 作用总结：防止session被误杀，定期清理过时的task

实现Delayed接口的模型pojo：

@AllArgsConstructor
@NoArgsConstructor
@Data
public class WebSocketTimeOut implements Delayed {

    private String sessionId;

    private Long expireTime;

    private Long version;

    @Override
    public long getDelay(@NotNull TimeUnit unit) {
        return unit.convert(expireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(@NotNull Delayed o) {
        return Long.compare(this.getDelay(TimeUnit.MILLISECONDS),
                o.getDelay(TimeUnit.MILLISECONDS));
    }
}

Session管理类：

/**
 * 创建一个管理session的管理器 全局单例 成员属性线程安全
 * 这只适合单机的服务器，如果是分布式集群则需要分布式redis
 */
@Component
@Slf4j
public class WebSocketSessionManage {

    /**
     * 初始化一个额外的线程去阻塞提取DelayQueue的任务 不然DelayQueue会阻塞当前的websocket业务线程
     */
    @PostConstruct
    public void initTimeOutCheck() {
        Thread thread = new Thread(this::timeOutWaitCheck);
        thread.setDaemon(true);
        thread.start();
    }

    /**
     * 定义超时时间 默认是毫秒
     */
    private static final long TIMEOUT = 300_00;

    /**
     * 存储sessionId-session映射集合
     */
    private final ConcurrentHashMap<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();

    /**
     * 存储每个队列内的task任务的版本号
     */
    private final ConcurrentHashMap<String, AtomicLong> versions = new ConcurrentHashMap<>();

    /**
     * 延迟队列
     */
    private final DelayQueue<WebSocketTimeOut> delayQueue = new DelayQueue<>();

    /**
     * 注册一个session
     *
     * @param session 连接信息
     */
    public void register(WebSocketSession session) {
        //放入session映射集合
        sessionMap.put(session.getId(), session);
        //新增版本信息
        versions.put(session.getId(), new AtomicLong(0));
        //放入队列 变为一个新的task 即使是同一个session
        refreshSessionTimeOut(session.getId());
    }

    /**
     * 刷新session的心跳
     *
     * @param id session的id
     */
    public void refreshSessionTimeOut(String id) {
        AtomicLong v = versions.get(id);

        if (v == null) {
            return;
        }

        long version = v.incrementAndGet(); //构造新的task的version
        long expire = System.currentTimeMillis() + TIMEOUT; //构造过期时间倒计时

        //放入队列
        delayQueue.offer(new WebSocketTimeOut(id, expire, version));
    }

    /**
     * 循环阻塞读取DelayQueue的任务进行过期处理
     */
    private void timeOutWaitCheck() {
        try {
            while (true) {
                //阻塞获取任务 他底层自动阻塞 因为全局单例 不用担心资源问题
                WebSocketTimeOut take = delayQueue.take();
                //处理过时连接
                handlerTimeOutTask(take);
            }
        } catch (InterruptedException e) {
            log.error("timeout handler error", e);
        }
    }

    /**
     * 处理过时连接
     *
     * @param take 任务
     */
    private void handlerTimeOutTask(WebSocketTimeOut take) {
        //检测旧任务
        String sessionId = take.getSessionId();
        AtomicLong v = versions.get(sessionId);
        if (v == null || v.get() != take.getVersion()) { //当前任务是旧任务 直接丢弃
            log.info("旧任务：sessionId:{}, version:{}", sessionId, take.getVersion());
            return;
        }

        //如果到这一步说明真的是假死用户、僵尸连接 直接干掉
        WebSocketSession webSocketSession = sessionMap.get(sessionId);
        if (webSocketSession == null || !webSocketSession.isOpen()) {
            return;
        }
        try {
            log.info("假死僵尸用户：sessionId:{}, version:{}", sessionId, take.getVersion());
            webSocketSession.close();
        } catch (IOException e) {
            log.warn("close websocket error, sessionId={}", sessionId, e);
        }

        remove(sessionId);
    }

    /**
     * 当连接已死 删除集合中的物品
     *
     * @param id sessionId
     */
    @SuppressWarnings("all")
    public void remove(String id) {
        sessionMap.remove(id);
        versions.remove(id);
    }
}

WebSocket实现类（重点接口）：

@Component
@Slf4j
public class HeartWebSocket extends TextWebSocketHandler {

    @Resource
    private WebSocketSessionManage webSocketSessionManage;

    /**
     * 客户端建立连接后
     *
     * @param session 代表客户端绑定的tcp连接信息
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("客户端连接信息: {}", session.getId());
        //新建连接
        webSocketSessionManage.register(session);
    }

    /**
     * 接收客户端发来的信息或心跳包
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        log.info("心跳信息: {}", message);
        //新建心跳task
        webSocketSessionManage.refreshSessionTimeOut(session.getId());
    }

    /**
     * 他是服务器给客户端发ping后收pong信息
     */
    @Override
    protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {
        super.handlePongMessage(session, message);
    }

    /**
     * 异常信息 报异常时触发
     */
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.info("异常信息: {}", exception.getMessage());
    }

    /**
     * 客户端关闭连接后
     *
     * @param session 代表客户端绑定的tcp连接信息
     * @param status  连接状态
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        webSocketSessionManage.remove(session.getId());
        log.info("连接信息: {}", session.isOpen());
        log.info("关闭状态: {}", status);
    }
}

● SSE

SSE本质上是客户端检测与服务器的连接健康性，这次换成服务器给客户端发心跳包了，而不是客户端给服务器，默认的SSE是业务上Http的长连接，就是一个连接一个线程，比如当一个sse请求（就是普通的controller请求和WebSocket不一样）来到时，自己手动new一个单线程的execute或者Runnable去while(true)维护这个长连接，这只是测试用的，真实肯定不这么干，而是创建一个Schedule因为Schedule是一个串行执行，每次执行都更新一个相同task给ThreadPoolTaskExecutor线程池（自己配置的，他默认是单线程，单线程也可以只不过如果某个任务阻塞了就炸线程了），每次SSE请求到来之后保存这个请求的SsEmitter，让Schedule遍历每一个SSE去给每个客户端发送心跳包来实现

○ Task

Task底层：

@Scheduled
    ↓
Spring 注册任务
    ↓
ThreadPoolTaskScheduler
    ↓
ScheduledExecutorService
    ↓
线程池执行

ScheduledExecutorService底层是DelayedQueue+wait+notify
也就是说他底层是DelayQueue就是任务重复执行就会把这个业务新建对象task放入queue不停地放不停地拿出

注：大部分可能是不new task而是复用一个task只更新计算时间，之所以说他是单线程是因为底层的ThreadPoolTaskExecutor默认的线程数是1但他还是个线程池

Task的简单使用：因为Task默认是单线程的，线程被阻塞可能就会卡主其他任务，所以要配置一个线程池ThreadPoolTaskSchedule

默认单线程流程：

• 开启EnableScheduling之后，spring底层会开启一个ThreadPoolTaskScheduler调度器（懒加载，当扫到第一个@Schedule之后才会初始化）
• 这个调度器底层是一个jdk的ScheduledExecutorService单线程模型
• 所有的@Schedule任务都会进入这个单线程串行执行
• 当一个先往DelayedQueue队列放入任务，执行完之后如果是循环按时间执行的话就更新重新计算这个业务的时间再次将这个task对象放入队列，这样的循环执行
• 如果是一次性任务执行完就直接丢弃

线程池流程：

• 开启EnableScheduling之后，spring底层会开启一个ThreadPoolTaskScheduler调度器（懒加载，当扫到第一个@Schedule之后才会初始化）
• 这时候初始化jdk底层的ScheduledThreadPoolExecutor线程池
• 所有的@Schedule都会进入DelayedQueue，而线程池的线程会从这个队列中取任务执行
• 形成单队列-多消费者模型，他底层其实是一个DelayedWorkdQueue是一个DelayedQueue的变种，这个队列是一个线程安全的队列

配置类：

@Configuration
@EnableScheduling
public class ScheduleConfig implements SchedulingConfigurer {

    /**
     * 设置线程池
     *
     * @param taskRegistrar 配置
     */
    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.setScheduler(getTaskScheduler());
    }

    /**
     * 创建线程池
     */
    @Bean
    public Executor getTaskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(5);                      // 核心线程数
        scheduler.setThreadNamePrefix("task-pool-");   // 线程名前缀
        scheduler.setAwaitTerminationSeconds(60);      // 等待终止时间
        scheduler.setWaitForTasksToCompleteOnShutdown(true); // 关机时等待任务完成
        scheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝策略
        scheduler.initialize();
        return scheduler;
    }
}

@Schedule的使用：

@Component
public class ScheduleTaskController {

    @Scheduled(cron = "0/10 * * * * ?") //每十秒执行一次 但是会被阻塞 为了测试多线程
    public void printer() {
        System.out.println("10s task...");
        try {
            Thread.sleep(1000000000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Scheduled(cron = "0/10 * * * * ?") //每十秒执行一次 正常任务
    public void printerPlus() {
        System.out.println("10s taskPlus...");
    }

    /**
     * fixedRate - 固定频率执行
     * 每 2 秒执行一次，不管上次是否完成
     * 注意：如果执行时间超过间隔，会立即开始下一次（依赖线程池大小）
     */
    @Scheduled(fixedRate = 2000) //2秒执行一次
    public void p3() {
        System.out.println("2s taskP3...");
    }

    /**
     * fixedDelay - 固定延迟执行
     * 上次执行完成后，延迟 3 秒再执行
     * 更安全，不会造成任务堆积
     */
    @Scheduled(fixedDelay = 3000)
    public void fixedDelayTask() {
        System.out.println("3s taskFixedDelay...");
    }

    /**
     * initialDelay - 首次延迟
     * 应用启动后，延迟 5 秒开始执行，然后每 5 秒执行一次
     */
    @Scheduled(initialDelay = 5000, fixedRate = 5000)
    public void initialDelayTask() {
        System.out.println("4s taskInitialDelay...");
    }
}

○ SSE使用搭配Schedule

如果设置了超时连接（new SseEmitter(30_000L)）：

SSE心跳检测的代码实现：

@Component
@Slf4j
public class SsEmitterManage {

    /**
     * 管理SSE的线程安全集合
     */
    public final ConcurrentHashMap<Long, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    /**
     * 注册一个连接的SSE
     *
     * @param key        连接key
     * @param sseEmitter 连接信息
     */
    public void registerSseEmitter(Long key, SseEmitter sseEmitter) {
        if (sseEmitterMap.containsKey(key)) { //同一个用户重新连接时重置sse
            sseEmitterMap.replace(key, sseEmitter);
        }else {
            sseEmitterMap.put(key, sseEmitter);
        }
        //注册事件 并不是真正执行 而是触发了某个条件后才执行lambda
        sseEmitter.onCompletion(() -> {
            sseEmitterMap.remove(key);
            log.info("sseEmitterMap remove key:{}", key);
        });
        sseEmitter.onTimeout(() -> sseEmitterMap.remove(key));
        sseEmitter.onError(e -> sseEmitterMap.remove(key));
    }

    /**
     * 循环发送SSE
     */
    @Scheduled(fixedDelay = 5000)
    public void whileSend() {
        sseEmitterMap.forEach( (key, sseEmitter) -> {
            try {
                log.info("sseEmitterMap:{}", sseEmitter);
                sseEmitter.send(SseEmitter.event().name("heartbeat").data("ping"));
            } catch (IOException e) {
                log.error(e.getMessage());
            }
        });
    }
}

SSE接口层面实现：

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

    @Resource
    private SsEmitterManage sseEmitterManage;

		//这里一定要设置produces = MediaType.TEXT_EVENT_STREAM_VALUE因为前端会检测返回值如果是json就会报异常
    @GetMapping(value = "/connect/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect(@PathVariable(value = "userId") Long userId) {
        SseEmitter sseEmitter = new SseEmitter(0L); //30秒重连，如果服务器给客户端发消息就会重置倒计时
        sseEmitterManage.registerSseEmitter(userId, sseEmitter);

        try {
            sseEmitter.send(SseEmitter.event()
                    .name("connect")
                    .data("SSE 连接已建立 - 用户ID: " + userId)
                    .reconnectTime(5000L)); // 5秒重连时间
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        log.info("🔗 SSE 连接已建立 - 用户ID: {}", userId);

        return sseEmitter;
    }
}

想要使用SSE需要额外配置Security的白名单，这样做会有个缺点就是这个请求并不能访问登录的用户信息，因为SecurityContextHoder上下文是Anonymous

auth.requestMatchers("/login", "/register", "/captchaImage", "/user/wechat/login", "/wx", "/sse/**").permitAll()

如果需要使用Security的上下文可以自己构造一个过滤器，就像WebSocket的握手拦截器一样：

@Component
@Order(Ordered.HIGHEST_PRECEDENCE) // 确保在 Security 过滤器链之前执行
public class SseHeaderFilter extends OncePerRequestFilter {
    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
            throws ServletException, IOException {
        
        String token = request.getParameter("token");
        String uri = request.getRequestURI();

        // 如果是 SSE 请求且带了 token
        if (uri.startsWith("/sse/") && StringUtils.hasText(token)) {
            // 使用 HttpServletRequestWrapper 包装请求，手动添加 Authorization Header
            HttpServletRequest wrapper = new HttpServletRequestWrapper(request) {
                @Override
                public String getHeader(String name) {
                    if ("Authorization".equalsIgnoreCase(name)) {
                        return "Bearer " + token;
                    }
                    return super.getHeader(name);
                }
            };
            filterChain.doFilter(wrapper, response);
            return;
        }
        
        filterChain.doFilter(request, response);
    }
}

● 项目笔记