jiujian / client-customer-order-api

Go to a project
Toggle navigation Toggle navigation pinning
zhuyifan
c3b93742 1 parent 1ebb8cf5

fix

Inline Side-by-side
Showing 1 changed file with 52 additions and 44 deletions
...@@ -7,14 +7,17 @@ import com.infoloop.tianting.server.session.UserSessionData; ...@@ -7,14 +7,17 @@ import com.infoloop.tianting.server.session.UserSessionData;
7 import com.infoloop.tianting.server.session.UserSessionKey; 7 import com.infoloop.tianting.server.session.UserSessionKey;
8 import com.infoloop.tianting.server.session.UserTypeEnum; 8 import com.infoloop.tianting.server.session.UserTypeEnum;
9 import lombok.extern.slf4j.Slf4j; 9 import lombok.extern.slf4j.Slf4j;
10 +import org.apache.commons.lang3.StringUtils;
10 import org.springframework.web.socket.CloseStatus; 11 import org.springframework.web.socket.CloseStatus;
11 import org.springframework.web.socket.TextMessage; 12 import org.springframework.web.socket.TextMessage;
12 import org.springframework.web.socket.WebSocketSession; 13 import org.springframework.web.socket.WebSocketSession;
13 import org.springframework.web.socket.handler.TextWebSocketHandler; 14 import org.springframework.web.socket.handler.TextWebSocketHandler;
14 15
16 +import javax.annotation.Nullable;
15 import java.io.IOException; 17 import java.io.IOException;
16 import java.util.List; 18 import java.util.List;
17 import java.util.concurrent.ConcurrentHashMap; 19 import java.util.concurrent.ConcurrentHashMap;
20 +import java.util.concurrent.CopyOnWriteArrayList;
18 import java.util.concurrent.Executors; 21 import java.util.concurrent.Executors;
19 import java.util.concurrent.ScheduledExecutorService; 22 import java.util.concurrent.ScheduledExecutorService;
20 import java.util.concurrent.TimeUnit; 23 import java.util.concurrent.TimeUnit;
...@@ -23,7 +26,7 @@ import java.util.stream.Collectors; ...@@ -23,7 +26,7 @@ import java.util.stream.Collectors;
23 @Slf4j 26 @Slf4j
24 public class WebSocketServer extends TextWebSocketHandler { 27 public class WebSocketServer extends TextWebSocketHandler {
25 28
26 - private static final ConcurrentHashMap<UserSessionKey, UserSessionData> userSessions = new ConcurrentHashMap<>(); 29 + private static final ConcurrentHashMap<UserSessionKey, List<UserSessionData>> userSessions = new ConcurrentHashMap<>();
27 private static final long INACTIVITY_TIMEOUT = 60 * 60 * 1000; 30 private static final long INACTIVITY_TIMEOUT = 60 * 60 * 1000;
28 private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); 31 private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
29 32
...@@ -33,38 +36,47 @@ public class WebSocketServer extends TextWebSocketHandler { ...@@ -33,38 +36,47 @@ public class WebSocketServer extends TextWebSocketHandler {
33 36
34 private static void cleanInactiveSessions() { 37 private static void cleanInactiveSessions() {
35 final var currentTime = System.currentTimeMillis(); 38 final var currentTime = System.currentTimeMillis();
36 - final var iterator = userSessions.entrySet().iterator(); 39 + userSessions.forEach((key, sessionList) ->
37 - log.info("cleanInactiveSessions; sessions:{}", userSessions.keySet()); 40 + userSessions.computeIfPresent(key, (k, sessions) -> {
38 - while (iterator.hasNext()) { 41 + sessions.removeIf(userData -> {
39 - final var entry = iterator.next(); 42 + if (currentTime - userData.getLastActiveTime() > INACTIVITY_TIMEOUT) {
40 - final var key = entry.getKey();
41 - final var userData = entry.getValue();
42 - final var lastActiveTime = userData.getLastActiveTime();
43 - if (currentTime - lastActiveTime > INACTIVITY_TIMEOUT) {
44 try { 43 try {
45 userData.closeSession(); 44 userData.closeSession();
46 - log.info("closeSession，userId : {}, userType : {}", key.getUserId(), key.getUserType()); 45 + log.info("closeSession, userId: {}, userType: {}", key.getUserId(), key.getUserType());
47 } catch (IOException e) { 46 } catch (IOException e) {
48 - log.error("closeSession error，userId : {}, userType : {}", key.getUserId(), key.getUserType(), e); 47 + log.error("closeSession error, userId: {}, userType: {}", key.getUserId(), key.getUserType(), e);
49 - }
50 - iterator.remove();
51 } 48 }
49 + return true;
52 } 50 }
51 + return false;
52 + });
53 + return sessions.isEmpty() ? null : sessions; // 为空时移除
54 + })
55 + );
53 } 56 }
54 57
58 +
55 public static <T> void sendMessageToUser(UserSessionKey key, SocketMessage<T> socketMessage) throws IOException { 59 public static <T> void sendMessageToUser(UserSessionKey key, SocketMessage<T> socketMessage) throws IOException {
56 sendMessageToUsers(List.of(key), socketMessage); 60 sendMessageToUsers(List.of(key), socketMessage);
57 } 61 }
58 62
59 public static <T> void sendMessageToUsers(List<UserSessionKey> keys, SocketMessage<T> socketMessage) throws IOException { 63 public static <T> void sendMessageToUsers(List<UserSessionKey> keys, SocketMessage<T> socketMessage) throws IOException {
60 for (final var key : keys) { 64 for (final var key : keys) {
61 - final var userData = userSessions.get(key); 65 + final var sessionList = userSessions.get(key);
62 - if (userData != null && userData.getSession().isOpen()) { 66 + if (sessionList != null) {
67 + sessionList.forEach(userData -> {
68 + try {
69 + if (userData.getSession().isOpen()) {
63 userData.getSession().sendMessage(new TextMessage(socketMessage.toJsonString())); 70 userData.getSession().sendMessage(new TextMessage(socketMessage.toJsonString()));
64 userData.updateLastActiveTime(); 71 userData.updateLastActiveTime();
65 - log.info("send message to userId : {}, userType : {}", key.getUserId(), key.getUserType()); 72 + log.info("Send message to userId: {}, userType: {}", key.getUserId(), key.getUserType());
66 } else { 73 } else {
67 - log.info("userId : {}, userType : {} WebSocket connect closed; ", key.getUserId(), key.getUserType()); 74 + log.info("WebSocket closed; userId: {}, userType: {}", key.getUserId(), key.getUserType());
75 + }
76 + } catch (IOException e) {
77 + log.error("Failed to send message; userId: {}, userType: {}", key.getUserId(), key.getUserType(), e);
78 + }
79 + });
68 } 80 }
69 } 81 }
70 } 82 }
...@@ -75,39 +87,40 @@ public class WebSocketServer extends TextWebSocketHandler { ...@@ -75,39 +87,40 @@ public class WebSocketServer extends TextWebSocketHandler {
75 87
76 public static <T> void sendMessageByUserTypes(List<UserTypeEnum> userTypes, SocketMessage<T> socketMessage) throws IOException { 88 public static <T> void sendMessageByUserTypes(List<UserTypeEnum> userTypes, SocketMessage<T> socketMessage) throws IOException {
77 final var keys = userSessions.keySet().stream() 89 final var keys = userSessions.keySet().stream()
78 - .filter(userSessionData -> userTypes.contains(userSessionData.getUserType())) 90 + .filter(userKey -> userTypes.contains(userKey.getUserType()))
79 .collect(Collectors.toList()); 91 .collect(Collectors.toList());
80 sendMessageToUsers(keys, socketMessage); 92 sendMessageToUsers(keys, socketMessage);
81 } 93 }
82 94
83 public static <T> void sendMessageByUserTypes(List<UserTypeEnum> userTypes, List<UserAuthData> userAuthData, SocketMessage<T> socketMessage) throws IOException { 95 public static <T> void sendMessageByUserTypes(List<UserTypeEnum> userTypes, List<UserAuthData> userAuthData, SocketMessage<T> socketMessage) throws IOException {
84 final var keys = userSessions.keySet().stream() 96 final var keys = userSessions.keySet().stream()
85 - .filter(userSessionData -> userTypes.contains(userSessionData.getUserType())) 97 + .filter(userKey -> userTypes.contains(userKey.getUserType()))
86 - .filter(userSessionData -> userAuthData.isEmpty() || (userAuthData.stream().anyMatch(data -> String.valueOf(data.getOperatorId()).equals(userSessionData.getUserId()) && data.getUserType().equals(userSessionData.getUserType())))) 98 + .filter(userKey -> userAuthData.isEmpty()
99 + || userAuthData.stream().anyMatch(data -> String.valueOf(data.getOperatorId()).equals(userKey.getUserId()) && data.getUserType().equals(userKey.getUserType())))
87 .collect(Collectors.toList()); 100 .collect(Collectors.toList());
88 sendMessageToUsers(keys, socketMessage); 101 sendMessageToUsers(keys, socketMessage);
89 } 102 }
90 103
91 public static void closeConnection(UserSessionKey key) { 104 public static void closeConnection(UserSessionKey key) {
92 - final var userData = userSessions.get(key); 105 + userSessions.computeIfPresent(key, (k, sessionList) -> {
93 - if (userData != null) { 106 + sessionList.forEach(userData -> {
94 try { 107 try {
95 userData.closeSession(); 108 userData.closeSession();
96 - userSessions.remove(key);
97 - log.info("WebSocket close; userId = {}, userType = {}", key.getUserId(), key.getUserType());
98 } catch (IOException e) { 109 } catch (IOException e) {
99 - log.error("WebSocket close failed: userId = {}, userType = {}", key.getUserId(), key.getUserType(), e); 110 + log.error("WebSocket close failed: userId: {}, userType: {}", key.getUserId(), key.getUserType(), e);
100 - }
101 - } else {
102 - log.warn("Not Found userId : {}, userType : {} WebSocket Connect ", key.getUserId(), key.getUserType());
103 } 111 }
112 + });
113 + return null; // 直接返回 null，移除 key
114 + });
115 + log.info("WebSocket closed; userId: {}, userType: {}", key.getUserId(), key.getUserType());
104 } 116 }
105 117
118 + @Nullable
106 private UserSessionKey getUserSessionKey(WebSocketSession session) { 119 private UserSessionKey getUserSessionKey(WebSocketSession session) {
107 final var userId = (String) session.getAttributes().get(CommonConstants.USER_ID); 120 final var userId = (String) session.getAttributes().get(CommonConstants.USER_ID);
108 final var userTypeStr = (String) session.getAttributes().get(CommonConstants.USER_TYPE); 121 final var userTypeStr = (String) session.getAttributes().get(CommonConstants.USER_TYPE);
109 final var userType = UserTypeEnum.fromString(userTypeStr); 122 final var userType = UserTypeEnum.fromString(userTypeStr);
110 - if (userId != null && userType != null) { 123 + if (StringUtils.isNotEmpty(userId) && userType != null) {
111 return UserSessionKey.builder().userId(userId).userType(userType).build(); 124 return UserSessionKey.builder().userId(userId).userType(userType).build();
112 } 125 }
113 return null; 126 return null;
...@@ -117,30 +130,25 @@ public class WebSocketServer extends TextWebSocketHandler { ...@@ -117,30 +130,25 @@ public class WebSocketServer extends TextWebSocketHandler {
117 public void afterConnectionEstablished(WebSocketSession session) { 130 public void afterConnectionEstablished(WebSocketSession session) {
118 final var key = getUserSessionKey(session); 131 final var key = getUserSessionKey(session);
119 if (key != null) { 132 if (key != null) {
120 - final var existingSession = userSessions.remove(key); 133 + userSessions.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>()).add(new UserSessionData(session));
121 - if (existingSession != null) { 134 + log.info("WebSocket connection established; userId: {}, userType: {}", key.getUserId(), key.getUserType());
122 - try {
123 - existingSession.closeSession();
124 - log.info("WebSocket old connect closed，userId : {}, userType : {}", key.getUserId(), key.getUserType());
125 - } catch (IOException e) {
126 - log.error("old WebSocket close failed，userId : {}, userType : {}", key.getUserId(), key.getUserType(), e);
127 - }
128 - }
129 - userSessions.put(key, new UserSessionData(session));
130 - log.info("WebSocket connect success, userId : {}, userType : {}", key.getUserId(), key.getUserType());
131 } else { 135 } else {
132 - log.warn("WebSocket connect failed，unable to get valid user information"); 136 + log.warn("WebSocket connection failed; unable to retrieve valid user information");
133 } 137 }
134 } 138 }
135 139
140 +
136 @Override 141 @Override
137 public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { 142 public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
138 final var key = getUserSessionKey(session); 143 final var key = getUserSessionKey(session);
139 if (key != null) { 144 if (key != null) {
140 - userSessions.remove(key); 145 + userSessions.computeIfPresent(key, (k, sessionList) -> {
141 - log.info("WebSocket closed: userId : {}, userType : {}", key.getUserId(), key.getUserType()); 146 + sessionList.removeIf(userData -> userData.getSession().getId().equals(session.getId()));
147 + return sessionList.isEmpty() ? null : sessionList;
148 + });
149 + log.info("WebSocket closed; userId: {}, userType: {}", key.getUserId(), key.getUserType());
142 } else { 150 } else {
143 - log.warn("WebSocket closed failed，unable to get valid user information"); 151 + log.warn("WebSocket closed; unable to retrieve valid user information");
144 } 152 }
145 } 153 }
146 } 154 }
...\ No newline at end of file ...\ No newline at end of file
......