Springboot 整合 Websocket 轻松实现IM及时通讯
一、方案实践
集成分为三步：添加依赖、增加配置类和消息核心类、前端集成。
1.1、添加依赖
org.springframework.boot
spring-boot-starter-websocket
2.1.13.RELEASE
1.2、增加WebSocket配置类
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* WebSocket配置
*/
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
1.3、增加消息核心类WebSocketServer
@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
// 消息存储
private static MessageStore messageStore;
// 消息发送
private static MessageSender messageSender;
public static void setMessageStore(MessageStore messageStore) {
WebSocketServer.messageStore = messageStore;
}
public static void setMessageSender(MessageSender messageSender) {
WebSocketServer.messageSender = messageSender;
}
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) {
messageStore.saveSession(session);
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose(Session session, @PathParam("userId") String userId) {
messageStore.deleteSession(session);
}
/**
* 收到客户端消息后调用的方法
*
* @ Param message 客户端发送过来的消息
*/
@OnMessage
public void onMessage(String message, Session session) throws Exception {
log.warn("=========== 收到来自窗口" + session.getId() + "的信息:" + message);
handleTextMessage(session, new TextMessage(message));
}
/**
* @param session
* @param error
*/
@OnError
public void onError(Session session, @PathParam("userId") String userId, Throwable error) {
log.error("=========== 发生错误");
error.printStackTrace();
// msgStore.deleteSession(session);
}
public void handleTextMessage(Session session, TextMessage message) throws Exception {
log.warn("=========== Received message: {}", message.getPayload());
}
}
1.4、前端页面加入socket
WebSocket Example
登录用户ID：
接受用户ID：
发送消息内容：
接受消息内容：
<script>
var socket = new WebSocket("ws://localhost:8080/websocket/aaa");
var roomId = "123456";
// 随机产出六位数字
var sendUserId = Math.floor(Math.random() * 1000000);
document.getElementById("sendUserId").value = sendUserId;
var messageReceive = document.getElementById("messageReceive");
socket.onopen = function (event) {
console.log("WebSocket is open now.");
let loginInfo = {
msgType: 2, //登录消息
sendUserId: sendUserId,
bizType: 1, //业务类型
bizOptModule: 1, //业务模块
roomId: roomId,
msgBody: {},
};
socket.send(JSON.stringify(loginInfo));
};
socket.onmessage = function (event) {
var message = event.data;
console.log("Received message: " + message);
messageReceive.value = message;
};
socket.onclose = function (event) {
console.log("WebSocket is closed now.");
};
function sendMessage() {
var message = document.getElementById("messageInput").value;
var receivedUserId = document.getElementById("receivedUserId").value;
let operateInfo = {
msgType: 100, //业务消息
sendUserId: sendUserId,
bizType: 1, //业务类型
bizOptModule: 1, //业务模块
roomId: roomId,
receivedUserId: receivedUserId,
msgBody: {
operateType: 1, //操作类型：禁言
operateContent: "1",
},
};
socket.send(JSON.stringify(operateInfo));
}
setInterval(() => {
socket.send("ping");
}, 30000);
</script>
二、小型及时通讯包含的模块
以上只是集成了Websocket框架，实现了基本的全双工通信，服务器和客户端都可以同时发送和接收数据。要想实现一些小型完整的即时通讯，还需要具备以下几个核心模块。架构图如下：
2.1、架构图
2.2、消息对象模型
组织消息内容，比如消息类型、发送者用户ID、接受者用户ID、具体的消息体等。比如：
public class SocketMsg {
/**
* 消息类型：1心跳 2登录 3业务操作
*/
private Integer msgType;
/**
* 发送者用户ID
*/
private String sendUserId;
/**
* 接受者用户ID
*/
private String receivedUserId;
/**
* 业务类型
*/
private Integer bizType;
/**
* 业务操作模块
*/
private Integer bizOptModule;
/**
* 消息内容
*/
private T msgBody;
}
2.3、消息存储模块
负责存储消息内容、用户ID和sessionID的关系，防止数据丢失或者服务器重启等。
2.4、消息发送模块
功能开发完毕，一般部署到分布式集群环境，所以通讯时session会分布在多台服务器。比如用户A的session在机器1，用户B的session在机器2，此时A发送给B，就无法单独通过机器1完成。
因为机器1拿不到机器2里的session，所以消息发不过去。此时只能借助别的中间件来实现，比如借助消息中间件kafka实现。
机器1将消息发送给kafka，然后机器1和机器2都监听kafka，然后查看用户对应的session是否在本机，如果在本机则发送出去。
2.5、消息推送模块
模块3提到的消息发送流程中，消息发送给 消息中间件，然后服务器消费到消费，再通过本机的session推送给客户端。
三、遇到的几个问题
3.1、连接自动断开
WebSocket连接之后，发现一个问题：每隔一段时间如果不传送数据的话，与前端的连接就会自动断开。可以采用心跳消息的方式来解决这个问题。比如客服端每隔30秒自动发送ping消息给服务端，服务端返回pong。
3.2、Session无法被序列化
分布式场景会存在这样的问题：当一次请求负载到第一台服务器时，session在第一台服务器线程上，第二次请求，负载到第二台服务器上，此时通过userId查找当前用户的session时，是查找不到的。
本来想着把session存入到redis中，就可以从redis获取用户的session，希望用这种方式来解决分布式场景下消息发送的问题。但是会出现如下错误：
The remote endpoint was in state [STREAM_WRITING] which is an invalid state for called method
翻看了session源码，发现session无法被序列化。所以这个方案只能放弃。解决方案请看下面的问题5或者上面的架构图。
3.3、对象无法自动注入
使用了@ServerEndpoint注解的类中使用@Resource或@Autowired注入对象都会失败，并且报空指针异常。
原因是WebSocket服务是线程安全的，那么当我们去发起一个ws连接时，就会创建一个端点对象。WebSocket服务是多对象的，不是单例的。而我们的Spring的Bean默认就是单例的，在非单例类中注入一个单例的Bean是冲突的。
或者说：
Spring管理采用单例模式（singleton），而 WebSocket 是多对象的，即每个客户端对应后台的一个 WebSocket 对象，也可以理解成 new 了一个 WebSocket，这样当然是不能获得自动注入的对象了，因为这两者刚好冲突。
@Autowired 注解注入对象操作是在启动时执行的，而不是在使用时，而 WebSocket 是只有连接使用时才实例化对象，且有多个连接就有多个对象。所以我们可以得出结论，这个 Service 根本就没有注入到 WebSocket 当中。
如何解决呢？
使用静态对象，并且对外暴露set方法，这样在对象初始化的时候，将其注入到WebSocketServer中。比如说这样：
@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
private static MessageStore messageStore;
private static MessageSender messageSender;
public static void setMessageStore(MessageStore messageStore) {
WebSocketServer.messageStore = messageStore;
}
public static void setMessageSender(MessageSender messageSender) {
WebSocketServer.messageSender = messageSender;
}
}
@Slf4j
@Service
public class MessageStore {
@Autowired
private RedisTemplate redisTemplate;
@PostConstruct
public void init() {
WebSocketServer.setMessageStore(this);
}
}
3.4、分布式场景消息如何发给客户端
问题2中提到了分布式场景下存在的session不在本机的问题，这种场景可以通过发送消息中间件的方式解决。具体这样解决：
每次连接时，都将userId和对应的session存入到本机，发送消息时，直接发送给MQ-Broker，然后每台应用负载都去消费这个消息，拿到消息之后，判断在本机能根据userId是否能找到session，找到session则推送到客户端。
3.5、部署时Nginx配置问题
代码开发完毕之后，本机跑通后，然后部署到服务器之后，还差很重要的一步：配置nginx代理。
3.5.1、给后端应用部署独立域名
要给后端应用部署独立域名，nginx代理直接转发到应用的独立域名，不要走微服务的gateway网关转发过去。
3.5.2、多层nginx转发问题
当只有一层nginx的时候，配置比较简单，如下：
location ~* ^/api/websocket/* {
proxy_pass http://mangodwsstest.mangod.top;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
proxy_set_header Host mangodwsstest.mangod.top;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "Upgrade";
proxy_set_header X-Real-IP $remote_addr;
}
但是，当有两层nginx转发的时候，问题就出现了。
在最外层的nginx需要使用如下配置，不能照抄后面一层的配置。proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for和proxy_set_header X-Forwarded-Proto $scheme这两个配置不能少，用来将协议和真实IP传递给后面一层的nginx。
location ~* ^/api/websocket/* {
proxy_pass http://mangodwsstest.mangod.top;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
proxy_set_header Host $http_host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
}
四、完整代码和示例
4.1、页面效果如下
开启两个web页面，用户1输入用户2的用户ID，输入发送消息内容，点击发送。在用户2的页面的接受消息内容可以看到发送的消息。
4.2、代码结构
4.3、代码地址
https://github.com/yclxiao/spring-websocket.git
五、总结
本文聊了Springboot如何集成Websocket、IM及时通讯需要哪些模块、开发和部署过程中遇到的问题、以及实现小型IM及时通讯的代码。希望对你有帮助！