背景
现在大多业务功能使用 socket.io实现长连接，但是部分第三方设备对接 只支持基础的websocket。
spring中使用基础的websocket, @OnMessage 收到消息，对消息的处理，if else 将会繁琐，难以维护。
本文仅介绍了如何使用enum枚举、java.util.function jdk8 函数式接口，实现消息的处理。
websocket 定义JSON 数据交换格式
本文使用的 示例格式：
//连接成功
{
"cmd":"connect","sn":"A7888","data":{
...}}
//设置人员
{
"cmd":"setUser","data":{
"userId":"1"}}
//控制设备 --多层级 的格式，第二层里面解析 仍可按照同样的方式来处理
{
"cmd":"to_client","data":{
"type":"openDoor","value":"ON"}}
springboot 集成websocket
pom.xml 依赖
<!-- spring websocket-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- spring websocket启动异常、排除spring-boot-starter-tomcat-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
</exclusion>
</exclusions>
</dependency>
定义WebSocketConfig
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
定义 @ServerEndpoint
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.websocket.OnOpen;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 定义websocket端点
*/
@Slf4j
@ServerEndpoint(value = "/socket/device")
@Component
public class DeviceServerEndpoint {
/**
* 记录当前在线连接数
*/
private static AtomicInteger onlineCount = new AtomicInteger(0);
/**
* 连接的对象
*/
public static final Map<String, Session> clientMap = new ConcurrentHashMap<>();
/**
* 收到客户端消息
*
* @param message 客户端发送过来的消息
* @throws
*/
@OnMessage
public void onMessage(String message, Session session) {
try {
DeviceMsg deviceMsg = JSON.parseObject(message, DeviceMsg.class);
if (deviceMsg != null && deviceMsg.getCmd() != null) {
// jdk8 函数式处理消息
deviceMsg.getCmd().consumer.accept(session, message);
} else {
log.info("无法自动处理，客户端消息：{}", message);
}
} catch (Exception e) {
log.error("消息处理失败", session.getId(), message);
e.printStackTrace();
}
}
/**
* 连接建立成功
*/
@OnOpen
public void onOpen(Session session) {
onlineCount.incrementAndGet(); // 在线数加1
log.info("有新连接加入：{}，当前在线数为：{}", session.getId(), onlineCount.get());
}
/**
* 连接关闭
*/
@OnClose
public void onClose(Session session) {
onlineCount.decrementAndGet(); // 在线数减1
log.info("有一连接关闭：{}，当前在线数为：{}", session.getId(), onlineCount.get());
}
@OnError
public void onError(Session session, Throwable error) {
onlineCount.decrementAndGet(); // 在线数减1
error.printStackTrace();
}
/**
* 测试 控制开锁
*/
public static void openDoor() {
//所有客户端发送消息
clientMap.forEach((id, session) -> {
session.getBasicRemote().sendText("ON");
}
}
deviceMsg实体类
import lombok.Data;
import java.io.Serializable;
@Data
public class DeviceMsg implements Serializable {
/**
* 指令
*/
Cmd cmd;
/**
* 数据块
*/
JSONObject data;
}
Cmd 核心消息处理 枚举类
import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import javax.websocket.Session;
import java.util.function.BiConsumer;
@Getter
@Slf4j
@AllArgsConstructor
public enum Cmd {
connect("设备连接成功", (Session session, String msg) -> {
//设备端连接成功，发送设置端消息，将Session记录起来
DeviceServerEndpoint.clientMap.put(session.getId(), session);
}),
ping("设备心跳", (Session session, String msg) -> {
session.getBasicRemote().sendText("pong");
}),
setUser("配置用户", (Session session, String msg) -> {
//拿到msg 转换对象或者其他操作
session.getBasicRemote().sendText("ok");
}),
to_client("客户端消息", (Session session, String msg) -> {
try {
String string = JSON.parseObject(msg).getJSONObject("data").getString("type");
Client clientCmd = Client.valueOf(string);
clientCmd.consumer.accept(session, msg);
} catch (Exception e) {
log.info("to_client客户端消息，无法自动处理：{}", msg);
}
});
/**
* 描述
*/
String desc;
/**
* 处理
*/
BiConsumer<Session, String> consumer;
}
Client 消息处理枚举类
import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import javax.websocket.Session;
import java.util.function.BiConsumer;
@Getter
@Slf4j
@AllArgsConstructor
public enum Client{
openDoor("客户端控制", (Session session, String msg) -> {
//测试 连接成功 直接 开门
DeviceServerEndpoint.openDoor();
}),
otherCmd("其他指令", (Session session, String msg) -> {
session.getBasicRemote().sendText("ok");
});
/**
* 描述
*/
String desc;
/**
* 处理
*/
BiConsumer<Session, String> consumer;
}
JDK8常用函数式编程接口介绍：
- Function<T, R>：接受一个类型为 T 的参数，返回类型为 R 的结果。常用方法包括 apply(T t)。
- Predicate：接受一个类型为 T 的参数，返回一个布尔值。常用方法包括 test(T t)。
- Consumer：接受一个类型为 T 的参数，没有返回值。常用方法包括 accept(T t)。
- Supplier：不接受任何参数，返回一个类型为 T 的结果。常用方法包括 get()。
- UnaryOperator：继承自 Function<T, R>，接受一个类型为 T 的参数，返回类型也为 T 的结果。常用方法包括 apply(T t)。
- BinaryOperator：继承自 BiFunction<T, U, R>，接受两个类型为 T 的参数，返回类型也为 T 的结果。常用方法包括 apply(T t1, T t2)。
- BiFunction<T, U, R>：接受两个参数，一个类型为 T，一个类型为 U，返回类型为 R 的结果。常用方法包括 apply(T t, U u)。
- BiPredicate<T, U>：接受两个参数，一个类型为 T，一个类型为 U，返回一个布尔值。常用方法包括 test(T t, U u)。
- BiConsumer<T, U> ：用于接受两个参数，一个类型为 T，一个类型为 U，并且没有返回值。
函数式编程接口的引入，使得在 Java 中能够更方便地实现函数式编程的特性，如Lambda表达式和方法引用。它们可以用于各种场景，例如集合的处理、条件判断、函数的组合等。通过使用这些接口，可以编写更简洁、可读性更高的代码。
附：
WebSocket介绍
