springboot 优雅使用函数式编程处理 websocket @OnMessage 消息

背景

现在大多业务功能使用 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 {@Beanpublic 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*/@OnMessagepublic 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();}}/*** 连接建立成功*/@OnOpenpublic void onOpen(Session session) {onlineCount.incrementAndGet(); // 在线数加1log.info("有新连接加入：{}，当前在线数为：{}", session.getId(), onlineCount.get());}/*** 连接关闭*/@OnClosepublic void onClose(Session session) {onlineCount.decrementAndGet(); // 在线数减1log.info("有一连接关闭：{}，当前在线数为：{}", session.getId(), onlineCount.get());}@OnErrorpublic void onError(Session session, Throwable error) {onlineCount.decrementAndGet(); // 在线数减1error.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，并且没有返回值。