位置：首页>> 软件编程>> java编程>> SpringBoot实现WebSocket即时通讯的示例代码

SpringBoot实现WebSocket即时通讯的示例代码

作者：码奴生来只知道前进~　　发布时间：2022-06-14 19:59:36　

标签：SpringBoot,WebSocket,即时通讯

1、引入依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.3</version>
</dependency>

2、WebSocketConfig 开启WebSocket

package com.shucha.deveiface.web.config;

/**
* @author tqf
* @Description
* @Version 1.0
* @since 2022-04-12 15:35
*/
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();
}
}

3、WebSocketServer

package com.shucha.deveiface.web.ws;

/**
* @author tqf
* @Description
* @Version 1.0
* @since 2022-04-12 15:33
*/
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@Component
@ServerEndpoint("/webSocket/{userId}")
@Slf4j
public class WebSocketServer {
private Session session;
private String userId;
/**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
private static int onlineCount = 0;
private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();

/**
* concurrent包的线程安全set，用来存放每个客户端对应的MyWebSocket对象
*/
private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap();

/**
* 为了保存在线用户信息，在方法中新建一个list存储一下【实际项目依据复杂度，可以存储到数据库或者缓存】
*/
private final static List<Session> SESSIONS = Collections.synchronizedList(new ArrayList<>());

/**
* 建立连接
* @param session
* @param userId
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) {
this.session = session;
this.userId = userId;
webSocketSet.add(this);
SESSIONS.add(session);
if (webSocketMap.containsKey(userId)) {
webSocketMap.remove(userId);
webSocketMap.put(userId,this);
} else {
webSocketMap.put(userId,this);
addOnlineCount();
}
// log.info("【websocket消息】有新的连接, 总数:{}", webSocketSet.size());
log.info("[连接ID:{}] 建立连接, 当前连接数:{}", this.userId, webSocketMap.size());
}

/**
* 断开连接
*/
@OnClose
public void onClose() {
webSocketSet.remove(this);
if (webSocketMap.containsKey(userId)) {
webSocketMap.remove(userId);
subOnlineCount();
}
// log.info("【websocket消息】连接断开, 总数:{}", webSocketSet.size());
log.info("[连接ID:{}] 断开连接, 当前连接数:{}", userId, webSocketMap.size());
}

/**
* 发送错误
* @param session
* @param error
*/
@OnError
public void onError(Session session, Throwable error) {
log.info("[连接ID:{}] 错误原因:{}", this.userId, error.getMessage());
error.printStackTrace();
}

/**
* 收到消息
* @param message
*/
@OnMessage
public void onMessage(String message) {
// log.info("【websocket消息】收到客户端发来的消息:{}", message);
log.info("[连接ID:{}] 收到消息:{}", this.userId, message);
}

/**
* 发送消息
* @param message
* @param userId
*/
public void sendMessage(String message,Long userId) {
WebSocketServer webSocketServer = webSocketMap.get(String.valueOf(userId));
if (webSocketServer!=null){
log.info("【websocket消息】推送消息, message={}", message);
try {
webSocketServer.session.getBasicRemote().sendText(message);
} catch (Exception e) {
e.printStackTrace();
log.error("[连接ID:{}] 发送消息失败, 消息:{}", this.userId, message, e);
}
}
}

/**
* 群发消息
* @param message
*/
public void sendMassMessage(String message) {
try {
for (Session session : SESSIONS) {
if (session.isOpen()) {
session.getBasicRemote().sendText(message);
log.info("[连接ID:{}] 发送消息:{}",session.getRequestParameterMap().get("userId"),message);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}

/**
* 获取当前连接数
* @return
*/
public static synchronized int getOnlineCount() {
return onlineCount;
}

/**
* 当前连接数加一
*/
public static synchronized void addOnlineCount() {
WebSocketServer.onlineCount++;
}

/**
* 当前连接数减一
*/
public static synchronized void subOnlineCount() {
WebSocketServer.onlineCount--;
}

}

4、测试连接发送和接收消息

package com.shucha.deveiface.web.controller;

import com.alibaba.fastjson.JSONObject;
import com.shucha.deveiface.web.ws.WebSocketServer;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
* @author tqf
* @Description
* @Version 1.0
* @since 2022-04-12 15:44
*/
@RestController
@RequestMapping("/web")
public class TestWebSocket {
@Autowired
private WebSocketServer webSocketServer;

/**
* 消息发送测试
*/
@GetMapping("/test")
public void test(){
for (int i=1;i<4;i++) {
WebsocketResponse response = new WebsocketResponse();
response.setUserId(String.valueOf(i));
response.setUserName("姓名"+ i);
response.setAge(i);
webSocketServer.sendMessage(JSONObject.toJSONString(response), Long.valueOf(String.valueOf(i)));
}
}

/**
* 群发消息测试(给当前连接用户发送)
*/
@GetMapping("/sendMassMessage")
public void sendMassMessage(){
WebsocketResponse response = new WebsocketResponse();
response.setUserName("群发消息模板测试");
webSocketServer.sendMassMessage(JSONObject.toJSONString(response));
}

@Data
@Accessors(chain = true)
public static class WebsocketResponse {
private String userId;
private String userName;
private int age;
}
}