1.引入依赖,(在微服务模块中)
<!-- Spring WebSocket -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>2.新建文件
package com.ruoyi.foundation.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 {
// 使用boot内置tomcat时需要注入此bean
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
package com.ruoyi.foundation.webSocket;
import lombok.extern.slf4j.Slf4j;
import javax.websocket.Session;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
public class WebsocketUtil {
private static final Map<String, Session> ONLINE_SESSION = new ConcurrentHashMap<>();
/**
* 添加session
*/
public static void addSession(String userId, Session session){
// 一个用户只允许一个session链接
ONLINE_SESSION.putIfAbsent(userId, session);
log.info("User [{}] connected. Total online users: {}", userId, ONLINE_SESSION.size());
}
/**
* 移除session
*/
public static void removeSession(String userId){
ONLINE_SESSION.remove(userId);
log.info("User [{}] disconnected. Total online users: {}", userId, ONLINE_SESSION.size());
}
/**
* 给单个用户推送消息
*/
public static void sendMessage(String userId, String message){
Session session = ONLINE_SESSION.get(userId);
if(session == null){
log.warn("Session for user [{}] not found", userId);
return;
}
sendMessage(session, message);
}
public static void sendMessage(Session session, String message) {
if (session != null) {
session.getAsyncRemote().sendText(message);
}
}
/**
* 给所有用户发消息
*/
public static void sendMessageForAll(String message) {
ONLINE_SESSION.forEach((userId, session) -> {
CompletableFuture.runAsync(() -> sendMessage(session, message))
.exceptionally(ex -> {
log.error("Failed to send message to user [{}]: {}", userId, ex.getMessage());
return null;
});
});
}
/**
* 给指定的多个用户推送消息
*/
public static void sendMessageForUsers(Set<String> userIds, String message) {
userIds.forEach(userId -> {
Session session = ONLINE_SESSION.get(userId);
if (session == null) {
log.warn("Session for user [{}] not found", userId);
return;
}
CompletableFuture.runAsync(() -> sendMessage(session, message))
.exceptionally(ex -> {
log.error("Failed to send message to user [{}]: {}", userId, ex.getMessage());
return null;
});
});
}
}
package com.ruoyi.foundation.apicontroller;
import com.google.gson.Gson;
import com.ruoyi.foundation.apicontroller.req.MemorialHallWebsocketReq;
import com.ruoyi.foundation.webSocket.WebsocketUtil;
import io.seata.common.util.StringUtils;
import org.apache.poi.util.StringUtil;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@Component
@ServerEndpoint(value = "/api/memorialHallWebsocket/{dailyMissId}/{userId}")
public class MmMemorialHallWebsocketController {
/**
* 保存每日一念纪念馆中当前在线的用户ID
*/
private static final Map<String, List<String>> memorialHallUsers = new ConcurrentHashMap<>();
private Gson gson=new Gson();
@OnOpen
public void onOpen(@PathParam(value = "dailyMissId") String dailyMissId,@PathParam(value = "userId") String userId, Session session) {
WebsocketUtil.addSession(userId, session);
List<String> strings = memorialHallUsers.get(dailyMissId);
if (strings == null){
List<String> list=new ArrayList<>();
list.add(userId);
memorialHallUsers.put(dailyMissId,list);
}else{
strings.add(userId);
}
}
@OnClose
public void onClose(@PathParam(value = "dailyMissId") String dailyMissId,@PathParam(value = "userId") String userId, Session session) {
WebsocketUtil.removeSession(userId);
List<String> strings = memorialHallUsers.get(dailyMissId);
if(strings != null){
strings.remove(userId);
}
}
@OnMessage
public void onMessage(@PathParam(value = "dailyMissId") String dailyMissId,@PathParam(value = "userId") String userId, Session session, String message) {
/*System.out.println(dailyMissId);
System.out.println(userId);
System.out.println(session);
System.out.println(message);*/
//MemorialHallWebsocketReq memorialHallWebsocketReq = gson.fromJson(message, MemorialHallWebsocketReq.class);
List<String> strings = memorialHallUsers.get(dailyMissId);
if(strings == null || strings.isEmpty()){
return;
}
Set<String> collect = strings.stream().filter(userId1 -> !StringUtils.equals(userId1, userId)).collect(Collectors.toSet());
//对同纪念馆的在线用户进行广播
WebsocketUtil.sendMessageForUsers(collect,message);
}
@OnError
public void onError(Session session, Throwable throwable) {
try {
session.close();
} catch (IOException e) {
e.printStackTrace();
}
throwable.printStackTrace();
}
}
3.网关允许WebScoket
- id: ruoyi-foundationWebSocket
uri: lb:ws://ruoyi-foundation
predicates:
- Path=/foundationWebSocket/**
filters:
- StripPrefix=1
4.测试
5.线上nginx配置
location /mmwzGateWay/ {
if ($request_method = OPTIONS) {
add_header Access-Control-Allow-Origin $http_origin;
add_header "Access-Control-Allow-Headers" "Authorization, Origin, X-Requested-With, Content-Type, Accept";
add_header Access-Control-Allow-Methods GET,POST,OPTIONS,HEAD,PUT,DELETE;
add_header Access-Control-Allow-Credentials true;
return 200;
}
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
# WebSocket 相关的头部配置
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header X-Forwarded-Proto $scheme;
proxy_pass http://mmwz-gateway:8080/;
#proxy_pass http://www.baidu.com/;
}
