Java Springboot SSE如何判断客户端能否正常接收消息-编程知识

背景

当新建一个 emitter 对象的时候, 它的默认超时时间是 30s.

SseEmitter emitter = new SseEmitter();

但是很多情况下, 默认30s的时间太短, 需要把 emitter 对象的超时时间设置成不超时, 也就是永久有效.

private static long DEFAULT_TIMEOUT = 0L;......SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);

这样也会带来一个问题, 就是永久有效的 emitter 对象如果没有调用关闭连接的接口的话 (比如用户直接关闭浏览器了) , 这个 emitter 对象就会一直存在.

解决方案

思路

sseEmitter 有下面的几个属性:

在这里插入图片描述

注意一下 sendFailed 这个属性, 我们可以利用这个属性来判断客户端能否正常接到消息.

当客户端无法接受消息时，SseEmitter对象在send一次之后sendFailed状态会变为True，这时候就可以剔除。同时在订阅时用此判断可以减少重复创建的机会

还有一个 complete 属性, 这个属性是与 sendFailed 有关的, 也就是消息发送成功的时候 complete 为 true, 失败的时候 complete 为 false. 我们可以用这个属性当做一个辅助.

请添加图片描述

拿到客户端是否能够正常接收消息这个状态以后, 我们就可以建立一个定时器，固定时间发送消息用来检测客户端是否离线.

代码

package com.example.demo.utils;import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.io.IOException;
import java.lang.reflect.Field;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;public class SSEUtils {public static Map<String, SseEmitter> subscribeMap = new ConcurrentHashMap<>();/**** 添加订阅* @param id 客户id* @return*/public static SseEmitter addSubscribe(String id) {SseEmitter sseEmitter = subscribeMap.get(id);if (sseEmitter == null) {sseEmitter = new SseEmitter(0L); // 永久有效sseEmitter.onTimeout(() -> {subscribeMap.remove(id);});sseEmitter.onError(throwable -> {subscribeMap.remove(id);});SseEmitter finalSseEmitter = sseEmitter;sseEmitter.onCompletion(() -> {subscribeMap.put(id, finalSseEmitter);});}return sseEmitter;}/**** 给单个用户发消息* @param id* @param msg* @return*/public static boolean sendSingleClientMsg(String id,Object msg) {SseEmitter sseEmitter = subscribeMap.get(id);if (sseEmitter == null) {return false;}try {sseEmitter.send(msg, MediaType.APPLICATION_JSON);return true;} catch (IOException e) {e.printStackTrace();return false;}}/**** 关闭订阅* @param id* @return*/public static boolean closeSubscribe(String id) {SseEmitter sseEmitter = subscribeMap.get(id);if (sseEmitter == null) {return true;}try {sseEmitter.complete();subscribeMap.remove(id);return true;} catch (Exception e) {e.printStackTrace();return false;}}/**** 检测客户端连接状态* @param sseEmitter* @return true代表还连接, false代表失去连接*/public static boolean checkSseConnectAlive(SseEmitter sseEmitter) {if (sseEmitter == null) {return false;}// 返回true代表还连接, 返回false代表失去连接return !(Boolean) getField(sseEmitter,sseEmitter.getClass(), "sendFailed") &&!(Boolean) getField(sseEmitter,sseEmitter.getClass(), "complete");}public static Object getField(Object obj, Class<?> clazz, String fieldName) {for (; clazz != Object.class; clazz = clazz.getSuperclass()) {try {Field field;field = clazz.getDeclaredField(fieldName);field.setAccessible(true);return field.get(obj);} catch (Exception e) {}}return null;}/**** 给所有客户端发消息* @param msg*/public void sendAllClientMsg(Object msg) {if (subscribeMap != null && !subscribeMap.isEmpty()) {for (String key : subscribeMap.keySet()) {// 发送检测消息sendSingleClientMsg(key,msg);// 判断客户端是否能接收到消息boolean isAlive = checkSseConnectAlive(subscribeMap.get(key));if (!isAlive) {// 断开连接的业务代码}}}}/**** 定时判断所有客户端状态*/@Async("threadPoolTaskExecutor")@Scheduled(fixedDelay = 1000*60*10) // 10minpublic void checkAlive() {sendAllClientMsg("CHECK_ALIVE");}
}

使用 @Scheduled 定时器, 不要忘记在启动类上面加这两个注解:

@SpringBootApplication
@EnableAsync
@EnableScheduling
public class DemoApplication {public static void main(String[] args) {SpringApplication.run(DemoApplication.class, args);}}

代码解释

重点部分是下面这段代码:

    /**** 检测客户端连接状态* @param sseEmitter* @return*/public static boolean checkSseConnectAlive(SseEmitter sseEmitter) {if (sseEmitter == null) {return false;}// 返回true代表还连接, 返回false代表失去连接return !(Boolean) getField(sseEmitter,sseEmitter.getClass(), "sendFailed") &&!(Boolean) getField(sseEmitter,sseEmitter.getClass(), "complete");}public static Object getField(Object obj, Class<?> clazz, String fieldName) {for (; clazz != Object.class; clazz = clazz.getSuperclass()) {try {Field field;field = clazz.getDeclaredField(fieldName);field.setAccessible(true);return field.get(obj);} catch (Exception e) {}}return null;}

1. 循环找 SseEmitter 和它的父类中是否存在 sendFailed 这个属性, 直到找到.

这是因为 sendFailed 这个属性是私有的, 不供外部访问, 这属性还正好在父类里, 所以要循环父类.

在这里插入图片描述

2. 通过 getDeclaredField() 方法拿到传入的 fieldName 的属性 (也就是 "sendFailed" 和 "complete" ), 接着使用 setAccessible(true) 把这个值设置为可访问的.

3. 最后通过 field.get(obj) 拿到这个属性的值, 也就是"sendFailed" 和 "complete" 的值是 true/false

思路和代码参考: Java Springboot SSE 解决永久存活判断客户端离线问题. 关于 SSE utils的一些工具类的方法在这个博客里面也有.

Java反射知识点补充

Java 反射是指在运行时动态地获取一个类的信息，并且可以操作它的属性、方法和构造方法等。Java 反射机制提供了一种在运行时检查、创建和操作对象的能力，这使得 Java 程序可以实现动态性和灵活性。

Java 反射机制主要包括以下三个类：

java.lang.Class 类：代表一个类，在运行时动态获取一个类的信息。
java.lang.reflect.Method 类：代表类的方法，在运行时可以使用 Method.invoke() 方法调用一个方法。
java.lang.reflect.Field 类：代表类的属性，在运行时可以使用 Field.get() 和 Field.set() 方法获取或设置一个属性的值。

以下是一个简单的 Java 反射示例，演示如何使用反射获取一个类的信息：

import java.lang.reflect.*;public class MyClass {private String name;private int age;public MyClass(String name, int age) {this.name = name;this.age = age;}public void sayHello() {System.out.println("Hello, " + name + "!");}public static void main(String[] args) throws Exception {// 获取 MyClass 类的 Class 对象Class<?> myClass = MyClass.class;// 创建一个 MyClass 对象MyClass obj = new MyClass("Bob", 20);// 获取 MyClass 类的构造方法，并使用它创建一个新的 MyClass 对象Constructor<?> constructor = myClass.getConstructor(String.class, int.class);MyClass newObj = (MyClass) constructor.newInstance("Alice", 30);// 获取 MyClass 类的属性，并使用它获取 obj 对象的 name 属性值Field field = myClass.getDeclaredField("name");field.setAccessible(true);String name = (String) field.get(obj);// 获取 MyClass 类的方法，并使用它调用 obj 对象的 sayHello 方法Method method = myClass.getMethod("sayHello");method.invoke(obj);System.out.println(name);         // 输出：BobSystem.out.println(newObj.name);  // 输出：Alice}
}