lywsvip/SpringBoot-Labs
1
0
Fork 0
mirror of https://gitee.com/yudaocode/SpringBoot-Labs.git synced 2026-09-03 05:53:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

增加 spring websocket 示例

Browse Source
This commit is contained in:
YunaiV
2019-11-21 00:56:45 +08:00
parent b4bd2067d0
commit 0806fbce26
19 changed files with 742 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lab-25/lab-websocket-25-01/pom.xml
View File

@@ -10,7 +10,7 @@
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>lab-25-01</artifactId>
<artifactId>lab-websocket-25-01</artifactId>
<dependencies>
<!-- 实现对 WebSocket 相关依赖的引入，方便~ -->

31
lab-25/lab-websocket-25-02/pom.xml Normal file
View File

@@ -0,0 +1,31 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.10.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>lab-websocket-25-02</artifactId>
<dependencies>
<!-- 实现对 WebSocket 相关依赖的引入，方便~ -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- 引入 Fastjson ，实现对 JSON 的序列化，因为后续我们会使用它解析消息 -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.62</version>
</dependency>
</dependencies>
</project>

13
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/Application.java Normal file
View File

@@ -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);
}
}

16
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/config/WebSocketConfiguration.java Normal file
View File

@@ -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();
}
}

40
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/AuthMessageHandler.java Normal file
View File

@@ -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<AuthRequest> {
@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;
}
}

25
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/MessageHandler.java Normal file
View File

@@ -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<T extends Message> {
/**
* 执行处理消息
*
* @param session 会话
* @param message 消息
*/
void execute(Session session, T message);
/**
* @return 消息类型，即每个 Message 实现类上的 TYPE 静态字段
*/
String getType();
}

32
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToAllHandler.java Normal file
View File

@@ -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<SendToAllRequest> {
@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;
}
}

32
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/handler/SendToOneHandler.java Normal file
View File

@@ -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<SendToOneRequest> {
@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;
}
}

24
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthRequest.java Normal file
View File

@@ -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;
}
}

37
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/AuthResponse.java Normal file
View File

@@ -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;
}
}

7
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/Message.java Normal file
View File

@@ -0,0 +1,7 @@
package cn.iocoder.springboot.lab25.springwebsocket.message;
/**
* 基础消息体
*/
public interface Message {
}

50
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendResponse.java Normal file
View File

@@ -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;
}
}

37
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToAllRequest.java Normal file
View File

@@ -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;
}
}

49
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToOneRequest.java Normal file
View File

@@ -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;
}
}

37
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/SendToUserRequest.java Normal file
View File

@@ -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;
}
}

24
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/message/UserJoinNoticeRequest.java Normal file
View File

@@ -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;
}
}

152
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/util/WebSocketUtil.java Normal file
View File

@@ -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, String> SESSION_USER_MAP = new ConcurrentHashMap<>();
/**
* 用户与 Session 的映射
*/
private static final Map<String, Session> 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 <T> 消息类型
*/
public static <T extends Message> 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 <T> 消息类型
*/
public static <T extends Message> void send(Session session, String type, T message) {
// 创建消息
String messageText = buildTextMessage(type, message);
// 遍历给单个 Session ，进行逐个发送
sendTextMessage(session, messageText);
}
/**
* 发送消息给指定用户
*
* @param user 指定用户
* @param type 消息类型
* @param message 消息体
* @param <T> 消息类型
* @return 发送是否成功你那个
*/
public static <T extends Message> 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 <T> 消息类型
* @return 消息
*/
private static <T extends Message> 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);
}
}
}

134
lab-25/lab-websocket-25-02/src/main/java/cn/iocoder/springboot/lab25/springwebsocket/websocket/WebsocketServerEndpoint.java Normal file
View File

@@ -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<String, MessageHandler> HANDLERS = new HashMap<>();
@Autowired
private ApplicationContext applicationContext;
@OnOpen
public void onOpen(Session session, EndpointConfig config) {
logger.info("[onOpen][session({}) 接入]", session);
// 解析 accessToken
List<String> accessTokenValues = session.getRequestParameterMap().get("accessToken");
String accessToken = !CollectionUtils.isEmpty(accessTokenValues) ? accessTokenValues.get(0) : null;
// 创建 AuthRequest 消息类型
AuthRequest authRequest = new AuthRequest().setAccessToken(accessToken);
// 获得消息处理器
MessageHandler<AuthRequest> 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<? extends Message> 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<? extends Message> 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<Message>) actualTypeArguments[0];
} else {
throw new IllegalStateException(String.format("类型(%s) 获得不到消息类型", handler));
}
}
}
}
}
throw new IllegalStateException(String.format("类型(%s) 获得不到消息类型", handler));
}
}

1
lab-25/pom.xml
View File

@@ -13,6 +13,7 @@
<packaging>pom</packaging>
<modules>
<module>lab-websocket-25-01</module>
<module>lab-websocket-25-02</module>
</modules>