From 0806fbce26f8056bd5fb740c1009a2ce3179680a Mon Sep 17 00:00:00 2001 From: YunaiV <> Date: Thu, 21 Nov 2019 00:56:45 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=20spring=20websocket=20?= =?UTF-8?q?=E7=A4=BA=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lab-25/lab-websocket-25-01/pom.xml | 2 +- lab-25/lab-websocket-25-02/pom.xml | 31 ++++ .../lab25/springwebsocket/Application.java | 13 ++ .../config/WebSocketConfiguration.java | 16 ++ .../handler/AuthMessageHandler.java | 40 +++++ .../handler/MessageHandler.java | 25 +++ .../handler/SendToAllHandler.java | 32 ++++ .../handler/SendToOneHandler.java | 32 ++++ .../springwebsocket/message/AuthRequest.java | 24 +++ .../springwebsocket/message/AuthResponse.java | 37 +++++ .../springwebsocket/message/Message.java | 7 + .../springwebsocket/message/SendResponse.java | 50 ++++++ .../message/SendToAllRequest.java | 37 +++++ .../message/SendToOneRequest.java | 49 ++++++ .../message/SendToUserRequest.java | 37 +++++ .../message/UserJoinNoticeRequest.java | 24 +++ .../springwebsocket/util/WebSocketUtil.java | 152 ++++++++++++++++++ .../websocket/WebsocketServerEndpoint.java | 134 +++++++++++++++ lab-25/pom.xml | 1 + 19 files changed, 742 insertions(+), 1 deletion(-) create mode 100644 lab-25/lab-websocket-25-02/pom.xml create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/Application.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/config/WebSocketConfiguration.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/AuthMessageHandler.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/MessageHandler.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToAllHandler.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToOneHandler.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthRequest.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthResponse.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/Message.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendResponse.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToAllRequest.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToOneRequest.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToUserRequest.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/UserJoinNoticeRequest.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/util/WebSocketUtil.java create mode 100644 lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/websocket/WebsocketServerEndpoint.java diff --git a/lab-25/lab-websocket-25-01/pom.xml b/lab-25/lab-websocket-25-01/pom.xml index 02682bae..3e105363 100644 --- a/lab-25/lab-websocket-25-01/pom.xml +++ b/lab-25/lab-websocket-25-01/pom.xml @@ -10,7 +10,7 @@ 4.0.0 - lab-25-01 + lab-websocket-25-01 diff --git a/lab-25/lab-websocket-25-02/pom.xml b/lab-25/lab-websocket-25-02/pom.xml new file mode 100644 index 00000000..832a7663 --- /dev/null +++ b/lab-25/lab-websocket-25-02/pom.xml @@ -0,0 +1,31 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.1.10.RELEASE + + + 4.0.0 + + lab-websocket-25-02 + + + + + org.springframework.boot + spring-boot-starter-websocket + + + + + com.alibaba + fastjson + 1.2.62 + + + + + diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/Application.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/Application.java new file mode 100644 index 00000000..a6f84bbf --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/Application.java @@ -0,0 +1,13 @@ +package cn.iocoder.springboot.lab25.springwebsocket; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/config/WebSocketConfiguration.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/config/WebSocketConfiguration.java new file mode 100644 index 00000000..47f36c44 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/config/WebSocketConfiguration.java @@ -0,0 +1,16 @@ +package cn.iocoder.springboot.lab25.springwebsocket.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.server.standard.ServerEndpointExporter; + +@Configuration +// @EnableWebSocket // 无需添加该注解，因为我们并不是使用 Spring WebSocket +public class WebSocketConfiguration { + + @Bean + public ServerEndpointExporter serverEndpointExporter() { + return new ServerEndpointExporter(); + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/AuthMessageHandler.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/AuthMessageHandler.java new file mode 100644 index 00000000..6d93efd4 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/AuthMessageHandler.java @@ -0,0 +1,40 @@ +package cn.iocoder.springboot.lab25.springwebsocket.handler; + +import cn.iocoder.springboot.lab25.springwebsocket.message.AuthRequest; +import cn.iocoder.springboot.lab25.springwebsocket.message.AuthResponse; +import cn.iocoder.springboot.lab25.springwebsocket.message.UserJoinNoticeRequest; +import cn.iocoder.springboot.lab25.springwebsocket.util.WebSocketUtil; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import javax.websocket.Session; + +@Component +public class AuthMessageHandler implements MessageHandler { + + @Override + public void execute(Session session, AuthRequest message) { + // 如果未传递 accessToken + if (StringUtils.isEmpty(message.getAccessToken())) { + WebSocketUtil.send(session, AuthResponse.TYPE, + new AuthResponse().setCode(1).setMessage("认证 accessToken 未传入")); + return; + } + + // 添加到 WebSocketUtil 中 + WebSocketUtil.addSession(session, message.getAccessToken()); // 考虑到代码简化，我们先直接使用 accessToken 作为 User + + // 判断是否认证成功。这里，假装直接成功 + WebSocketUtil.send(session, AuthResponse.TYPE, new AuthResponse().setCode(0)); + + // 通知所有人，某个人加入了。这个是可选逻辑，仅仅是为了演示 + WebSocketUtil.broadcast(UserJoinNoticeRequest.TYPE, + new UserJoinNoticeRequest().setNickname(message.getAccessToken())); // 考虑到代码简化，我们先直接使用 accessToken 作为 User + } + + @Override + public String getType() { + return AuthRequest.TYPE; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/MessageHandler.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/MessageHandler.java new file mode 100644 index 00000000..aa89c13c --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/MessageHandler.java @@ -0,0 +1,25 @@ +package cn.iocoder.springboot.lab25.springwebsocket.handler; + +import cn.iocoder.springboot.lab25.springwebsocket.message.Message; + +import javax.websocket.Session; + +/** + * 消息处理器接口 + */ +public interface MessageHandler { + + /** + * 执行处理消息 + * + * @param session 会话 + * @param message 消息 + */ + void execute(Session session, T message); + + /** + * @return 消息类型，即每个 Message 实现类上的 TYPE 静态字段 + */ + String getType(); + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToAllHandler.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToAllHandler.java new file mode 100644 index 00000000..cda1efac --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToAllHandler.java @@ -0,0 +1,32 @@ +package cn.iocoder.springboot.lab25.springwebsocket.handler; + +import cn.iocoder.springboot.lab25.springwebsocket.message.SendResponse; +import cn.iocoder.springboot.lab25.springwebsocket.message.SendToAllRequest; +import cn.iocoder.springboot.lab25.springwebsocket.message.SendToUserRequest; +import cn.iocoder.springboot.lab25.springwebsocket.util.WebSocketUtil; +import org.springframework.stereotype.Component; + +import javax.websocket.Session; + +@Component +public class SendToAllHandler implements MessageHandler { + + @Override + public void execute(Session session, SendToAllRequest message) { + // 这里，假装直接成功 + SendResponse sendResponse = new SendResponse().setMsgId(message.getMsgId()).setCode(0); + WebSocketUtil.send(session, SendResponse.TYPE, sendResponse); + + // 创建转发的消息 + SendToUserRequest sendToUserRequest = new SendToUserRequest().setMsgId(message.getMsgId()) + .setContent(message.getContent()); + // 广播发送 + WebSocketUtil.broadcast(SendToUserRequest.TYPE, sendToUserRequest); + } + + @Override + public String getType() { + return SendToAllRequest.TYPE; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToOneHandler.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToOneHandler.java new file mode 100644 index 00000000..d7437bd4 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToOneHandler.java @@ -0,0 +1,32 @@ +package cn.iocoder.springboot.lab25.springwebsocket.handler; + +import cn.iocoder.springboot.lab25.springwebsocket.message.SendResponse; +import cn.iocoder.springboot.lab25.springwebsocket.message.SendToOneRequest; +import cn.iocoder.springboot.lab25.springwebsocket.message.SendToUserRequest; +import cn.iocoder.springboot.lab25.springwebsocket.util.WebSocketUtil; +import org.springframework.stereotype.Component; + +import javax.websocket.Session; + +@Component +public class SendToOneHandler implements MessageHandler { + + @Override + public void execute(Session session, SendToOneRequest message) { + // 这里，假装直接成功 + SendResponse sendResponse = new SendResponse().setMsgId(message.getMsgId()).setCode(0); + WebSocketUtil.send(session, SendResponse.TYPE, sendResponse); + + // 创建转发的消息 + SendToUserRequest sendToUserRequest = new SendToUserRequest().setMsgId(message.getMsgId()) + .setContent(message.getContent()); + // 广播发送 + WebSocketUtil.send(message.getToUser(), SendToUserRequest.TYPE, sendToUserRequest); + } + + @Override + public String getType() { + return SendToOneRequest.TYPE; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthRequest.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthRequest.java new file mode 100644 index 00000000..29b8bf11 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthRequest.java @@ -0,0 +1,24 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 用户认证请求 + */ +public class AuthRequest implements Message { + + public static final String TYPE = "AUTH_REQUEST"; + + /** + * 认证 Token + */ + private String accessToken; + + public String getAccessToken() { + return accessToken; + } + + public AuthRequest setAccessToken(String accessToken) { + this.accessToken = accessToken; + return this; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthResponse.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthResponse.java new file mode 100644 index 00000000..cfa4fbda --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthResponse.java @@ -0,0 +1,37 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 用户认证响应 + */ +public class AuthResponse implements Message { + + public static final String TYPE = "AUTH_RESPONSE"; + + /** + * 响应状态码 + */ + private Integer code; + /** + * 响应提示 + */ + private String message; + + public Integer getCode() { + return code; + } + + public AuthResponse setCode(Integer code) { + this.code = code; + return this; + } + + public String getMessage() { + return message; + } + + public AuthResponse setMessage(String message) { + this.message = message; + return this; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/Message.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/Message.java new file mode 100644 index 00000000..974b146a --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/Message.java @@ -0,0 +1,7 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 基础消息体 + */ +public interface Message { +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendResponse.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendResponse.java new file mode 100644 index 00000000..c695cf83 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendResponse.java @@ -0,0 +1,50 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 发送消息响应结果的 Message + */ +public class SendResponse implements Message { + + public static final String TYPE = "SEND_RESPONSE"; + + /** + * 消息编号 + */ + private String msgId; + /** + * 响应状态码 + */ + private Integer code; + /** + * 响应提示 + */ + private String message; + + public String getMsgId() { + return msgId; + } + + public SendResponse setMsgId(String msgId) { + this.msgId = msgId; + return this; + } + + public Integer getCode() { + return code; + } + + public SendResponse setCode(Integer code) { + this.code = code; + return this; + } + + public String getMessage() { + return message; + } + + public SendResponse setMessage(String message) { + this.message = message; + return this; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToAllRequest.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToAllRequest.java new file mode 100644 index 00000000..f75e0f9b --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToAllRequest.java @@ -0,0 +1,37 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 发送给所有人的群聊消息的 Message + */ +public class SendToAllRequest implements Message { + + public static final String TYPE = "SEND_TO_ALL_REQUEST"; + + /** + * 消息编号 + */ + private String msgId; + /** + * 内容 + */ + private String content; + + public String getContent() { + return content; + } + + public SendToAllRequest setContent(String content) { + this.content = content; + return this; + } + + public String getMsgId() { + return msgId; + } + + public SendToAllRequest setMsgId(String msgId) { + this.msgId = msgId; + return this; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToOneRequest.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToOneRequest.java new file mode 100644 index 00000000..8daa74ca --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToOneRequest.java @@ -0,0 +1,49 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 发送给指定人的私聊消息的 Message + */ +public class SendToOneRequest implements Message { + + public static final String TYPE = "SEND_TO_ONE_REQUEST"; + + /** + * 发送给的用户 + */ + private String toUser; + /** + * 消息编号 + */ + private String msgId; + /** + * 内容 + */ + private String content; + + public String getToUser() { + return toUser; + } + + public SendToOneRequest setToUser(String toUser) { + this.toUser = toUser; + return this; + } + + public String getMsgId() { + return msgId; + } + + public SendToOneRequest setMsgId(String msgId) { + this.msgId = msgId; + return this; + } + + public String getContent() { + return content; + } + + public SendToOneRequest setContent(String content) { + this.content = content; + return this; + } +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToUserRequest.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToUserRequest.java new file mode 100644 index 00000000..67ad15c0 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToUserRequest.java @@ -0,0 +1,37 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 发送消息给一个用户的 Message + */ +public class SendToUserRequest implements Message { + + public static final String TYPE = "SEND_TO_USER_REQUEST"; + + /** + * 消息编号 + */ + private String msgId; + /** + * 内容 + */ + private String content; + + public String getMsgId() { + return msgId; + } + + public SendToUserRequest setMsgId(String msgId) { + this.msgId = msgId; + return this; + } + + public String getContent() { + return content; + } + + public SendToUserRequest setContent(String content) { + this.content = content; + return this; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/UserJoinNoticeRequest.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/UserJoinNoticeRequest.java new file mode 100644 index 00000000..e1e923af --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/UserJoinNoticeRequest.java @@ -0,0 +1,24 @@ +package cn.iocoder.springboot.lab25.springwebsocket.message; + +/** + * 用户加入群聊的通知 Message + */ +public class UserJoinNoticeRequest implements Message { + + public static final String TYPE = "USER_JOIN_NOTICE_REQUEST"; + + /** + * 昵称 + */ + private String nickname; + + public String getNickname() { + return nickname; + } + + public UserJoinNoticeRequest setNickname(String nickname) { + this.nickname = nickname; + return this; + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/util/WebSocketUtil.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/util/WebSocketUtil.java new file mode 100644 index 00000000..d135a1f7 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/util/WebSocketUtil.java @@ -0,0 +1,152 @@ +package cn.iocoder.springboot.lab25.springwebsocket.util; + +import cn.iocoder.springboot.lab25.springwebsocket.message.Message; +import com.alibaba.fastjson.JSONObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.websocket.RemoteEndpoint; +import javax.websocket.Session; +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * WebSocket 工具类，提供客户端连接的管理等功能 + */ +public class WebSocketUtil { + + private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUtil.class); + + // ========== 会话相关 ========== + + /** + * Session 与用户的映射 + */ + private static final Map SESSION_USER_MAP = new ConcurrentHashMap<>(); + /** + * 用户与 Session 的映射 + */ + private static final Map USER_SESSION_MAP = new ConcurrentHashMap<>(); + + /** + * 添加 Session 。在这个方法中，会添加用户和 Session 之间的映射 + * + * @param session Session + * @param user 用户 + */ + public static void addSession(Session session, String user) { + // 更新 USER_SESSION_MAP + USER_SESSION_MAP.put(user, session); + // 更新 SESSION_USER_MAP + SESSION_USER_MAP.put(session, user); + } + + /** + * 移除 Session 。 + * + * @param session Session + */ + public static void removeSession(Session session) { + // 从 SESSION_USER_MAP 中移除 + String user = SESSION_USER_MAP.remove(session); + // 从 USER_SESSION_MAP 中移除 + if (user != null && user.length() > 0) { + USER_SESSION_MAP.remove(user); + } + } + + // ========== 消息相关 ========== + + /** + * 广播发送消息给所有在线用户 + * + * @param type 消息类型 + * @param message 消息体 + * @param 消息类型 + */ + public static void broadcast(String type, T message) { + // 创建消息 + String messageText = buildTextMessage(type, message); + // 遍历 SESSION_USER_MAP ，进行逐个发送 + for (Session session : SESSION_USER_MAP.keySet()) { + sendTextMessage(session, messageText); + } + } + + /** + * 发送消息给单个用户的 Session + * + * @param session Session + * @param type 消息类型 + * @param message 消息体 + * @param 消息类型 + */ + public static void send(Session session, String type, T message) { + // 创建消息 + String messageText = buildTextMessage(type, message); + // 遍历给单个 Session ，进行逐个发送 + sendTextMessage(session, messageText); + } + + /** + * 发送消息给指定用户 + * + * @param user 指定用户 + * @param type 消息类型 + * @param message 消息体 + * @param 消息类型 + * @return 发送是否成功你那个 + */ + public static boolean send(String user, String type, T message) { + // 获得用户对应的 Session + Session session = USER_SESSION_MAP.get(user); + if (session == null) { + LOGGER.error("[send][user({}) 不存在对应的 session]", user); + return false; + } + // 发送消息 + send(session, type, message); + return true; + } + + /** + * 构建完整的消息 + * + * @param type 消息类型 + * @param message 消息体 + * @param 消息类型 + * @return 消息 + */ + private static String buildTextMessage(String type, T message) { + JSONObject messageObject = new JSONObject(); + messageObject.put("type", type); + messageObject.put("body", message); + return messageObject.toString(); + } + + /** + * 真正发送消息 + * + * @param session Session + * @param messageText 消息 + */ + private static void sendTextMessage(Session session, String messageText) { + if (session == null) { + LOGGER.error("[sendTextMessage][session 为 null]"); + return; + } + RemoteEndpoint.Basic basic = session.getBasicRemote(); + if (basic == null) { + LOGGER.error("[sendTextMessage][session 的 为 null]"); + return; + } + try { + basic.sendText(messageText); + } catch (IOException e) { + LOGGER.error("[sendTextMessage][session({}) 发送消息{}) 发生异常", + session, messageText, e); + } + } + +} diff --git a/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/websocket/WebsocketServerEndpoint.java b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/websocket/WebsocketServerEndpoint.java new file mode 100644 index 00000000..dde62c33 --- /dev/null +++ b/lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/websocket/WebsocketServerEndpoint.java @@ -0,0 +1,134 @@ +package cn.iocoder.springboot.lab25.springwebsocket.websocket; + +import cn.iocoder.springboot.lab25.springwebsocket.handler.MessageHandler; +import cn.iocoder.springboot.lab25.springwebsocket.message.AuthRequest; +import cn.iocoder.springboot.lab25.springwebsocket.message.Message; +import cn.iocoder.springboot.lab25.springwebsocket.util.WebSocketUtil; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.aop.framework.AopProxyUtils; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.stereotype.Controller; +import org.springframework.util.CollectionUtils; + +import javax.websocket.*; +import javax.websocket.server.ServerEndpoint; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +@Controller +@ServerEndpoint("/") +public class WebsocketServerEndpoint implements InitializingBean { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + /** + * 消息类型与 MessageHandler 的映射 + * + * 注意，这里设置成静态变量。虽然说 WebsocketServerEndpoint 是单例，但是 Spring Boot 还是会为每个 WebSocket 创建一个 WebsocketServerEndpoint Bean 。 + */ + private static final Map HANDLERS = new HashMap<>(); + + @Autowired + private ApplicationContext applicationContext; + + @OnOpen + public void onOpen(Session session, EndpointConfig config) { + logger.info("[onOpen][session({}) 接入]", session); + // 解析 accessToken + List accessTokenValues = session.getRequestParameterMap().get("accessToken"); + String accessToken = !CollectionUtils.isEmpty(accessTokenValues) ? accessTokenValues.get(0) : null; + // 创建 AuthRequest 消息类型 + AuthRequest authRequest = new AuthRequest().setAccessToken(accessToken); + // 获得消息处理器 + MessageHandler messageHandler = HANDLERS.get(AuthRequest.TYPE); + if (messageHandler == null) { + logger.error("[onOpen][认证消息类型，不存在消息处理器]"); + return; + } + messageHandler.execute(session, authRequest); + } + + @OnMessage + public void onMessage(Session session, String message) { + logger.info("[onOpen][session({}) 接收到一条消息({})]", session, message); // 生产环境下，请设置成 debug 级别 + try { + // 获得消息类型 + JSONObject jsonMessage = JSON.parseObject(message); + String messageType = jsonMessage.getString("type"); + // 获得消息处理器 + MessageHandler messageHandler = HANDLERS.get(messageType); + if (messageHandler == null) { + logger.error("[onMessage][消息类型({}) 不存在消息处理器]", messageType); + return; + } + // 解析消息 + Class messageClass = this.getMessageClass(messageHandler); + // 处理消息 + Message messageObj = JSON.parseObject(jsonMessage.getString("body"), messageClass); + messageHandler.execute(session, messageObj); + } catch (Throwable throwable) { + logger.info("[onMessage][session({}) message({}) 发生异常]", session, throwable); + } + } + + @OnClose + public void onClose(Session session, CloseReason closeReason) { + logger.info("[onClose][session({}) 连接关闭。关闭原因是({})}]", session, closeReason); + WebSocketUtil.removeSession(session); + } + + @OnError + public void onError(Session session, Throwable throwable) { + logger.info("[onClose][session({}) 发生异常]", session, throwable); + } + + @Override + public void afterPropertiesSet() throws Exception { + // 通过 ApplicationContext 获得所有 MessageHandler Bean + applicationContext.getBeansOfType(MessageHandler.class).values() // 获得所有 MessageHandler Bean + .forEach(messageHandler -> HANDLERS.put(messageHandler.getType(), messageHandler)); // 添加到 handlers 中 + logger.info("[afterPropertiesSet][消息处理器数量：{}]", HANDLERS.size()); + } + + private Class getMessageClass(MessageHandler handler) { + // 获得 Bean 对应的 Class 类名。因为有可能被 AOP 代理过。 + Class targetClass = AopProxyUtils.ultimateTargetClass(handler); + // 获得接口的 Type 数组 + Type[] interfaces = targetClass.getGenericInterfaces(); + Class superclass = targetClass.getSuperclass(); + while ((Objects.isNull(interfaces) || 0 == interfaces.length) && Objects.nonNull(superclass)) { // 此处，是以父类的接口为准 + interfaces = superclass.getGenericInterfaces(); + superclass = targetClass.getSuperclass(); + } + if (Objects.nonNull(interfaces)) { + // 遍历 interfaces 数组 + for (Type type : interfaces) { + // 要求 type 是泛型参数 + if (type instanceof ParameterizedType) { + ParameterizedType parameterizedType = (ParameterizedType) type; + // 要求是 MessageHandler 接口 + if (Objects.equals(parameterizedType.getRawType(), MessageHandler.class)) { + Type[] actualTypeArguments = parameterizedType.getActualTypeArguments(); + // 取首个元素 + if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) { + return (Class) actualTypeArguments[0]; + } else { + throw new IllegalStateException(String.format("类型(%s) 获得不到消息类型", handler)); + } + } + } + } + } + throw new IllegalStateException(String.format("类型(%s) 获得不到消息类型", handler)); + } + +} diff --git a/lab-25/pom.xml b/lab-25/pom.xml index b90adbb4..da81aa31 100644 --- a/lab-25/pom.xml +++ b/lab-25/pom.xml @@ -13,6 +13,7 @@ pom lab-websocket-25-01 + lab-websocket-25-02