1、新建WebSocketConfig配置类
package com.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }
2、新建具体业务socket
package com.socket; import lombok.extern.slf4j.Slf4j; 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.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArraySet; @ServerEndpoint("/sysMessageSocket/{token}") @Component @Slf4j public class SysMessageWebSocket { private Session session; private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>(); private static CopyOnWriteArraySet<SysMessageWebSocket> webSocketSet = new CopyOnWriteArraySet<>(); @OnOpen public void onOpen(Session session, @PathParam(value = "token") String token) { log.info("WebSocket建立连接中,连接用户ID:{}", token); try { Session historySession = sessionPool.get(token); if (historySession != null) { webSocketSet.remove(historySession); historySession.close(); } } catch (IOException e) { log.error("重复登录异常,错误信息:" + e.getMessage(), e); } this.session = session; webSocketSet.add(this); sessionPool.put(token, session); log.info("建立连接完成,当前在线人数为:{}", webSocketSet.size()); } @OnError public void one rror(Throwable throwable) { throwable.printStackTrace(); } @OnClose public void onClose() { webSocketSet.remove(this); String token = this.session.getRequestParameterMap().get("token").get(0); log.info(token + "连接断开,当前在线人数为:{}", webSocketSet.size()); } @OnMessage public void onMessage(String message) { log.info("收到客户端发来的消息:{}", message); } public void sendMessageByUser(String token, String message) { log.info("用户ID:" + token + ",推送内容:" + message); Session session = sessionPool.get(token); try { if(session!=null){ session.getBasicRemote().sendText(message); }else{ log.info("连接已断开,用户ID:" + token + ",内容推送失败:" + message + "。session为空。"); } } catch (IllegalStateException | IOException e) { log.info("连接已断开,用户ID:" + token + ",内容推送失败:" + message + "。"+e.getMessage()); } } public void sendAllMessage(String message) { log.info("发送消息:{}", message); for (SysMessageWebSocket webSocket : webSocketSet) { try { webSocket.session.getBasicRemote().sendText(message); } catch (IOException e) { log.error("群发消息发生错误:" + e.getMessage(), e); } } } }
3、测试
通过 www.jsons.cn/websocket/ 测试本地socket是否互通
注意:
config配置文件WebSocketConfig的ServerEndpointExporter(服务器端点导出器)与@Test冲突(websocket是需要依赖tomcat等容器的启动)
方案:本地启用测试代码时需要将@Bean注掉(提交git的代码需要将Test代码注掉);
新建测试类时,在springbootTest后添加内容webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT。
例如测试类为
package com.test; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest(classes = MicroenterpriseApplication.class,webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) public class VerifyCodeServiceImplTest { @Test public void getVerifyCode() { } }
标签:webSocket,springboot,token,session,org,import,message,public From: https://www.cnblogs.com/xiufengd/p/17083155.html