Springboot实现websocket

Posted 2022-11-02 诺浅

Websocket

在前面的文章中我们讲到了websocket和传统的http有什么区别，以及websocket用来实现服务器推的优势，那么这篇文章就来说说，在Java中如果要实现一个websocket应该怎么实现，这里采用的是更加方便的Springboot的方式，如果项目中没有使用springboot框架，也是可以实现的，本文不涉及。

代码

服务端

1、先引入springboot-websocket的starter

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2、定义一个WebSocketBean，用来保存组装的信息，这个Bean也可以自己添加其他的属性

@Getter
@Setter
public class WebSocketBean 
    private WebSocketSession webSocketSession;
    private int clientId;
    private String token;

3、连接或断开处理类

@Slf4j
public class MyWebsocketHandler extends AbstractWebSocketHandler 

    public static final Map<String, WebSocketBean> webSocketBeanMap;

    /**
     * 仅用用于标识客户端编号
     */
    private static final AtomicInteger clientIdMaker;

    static 
        webSocketBeanMap = new ConcurrentHashMap<>();
        clientIdMaker = new AtomicInteger(0);
    

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception 
        // 当WebSocket连接正式建立后，将该Session加入到Map中进行管理
        Map<String, Object> attributes = session.getAttributes();
        WebSocketBean webSocketBean = new WebSocketBean();
        webSocketBean.setWebSocketSession(session);
        webSocketBean.setClientId(clientIdMaker.getAndIncrement());
        webSocketBean.setToken(attributes.get("token").toString());
        webSocketBeanMap.put(session.getId(), webSocketBean);
    

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception 
        //当连接关闭后，从Map中移除session实例
        webSocketBeanMap.remove(session.getId());
    

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception 
        //传输过程中出现了错误
        if (session.isOpen()) 
            session.close();
        
        webSocketBeanMap.remove(session.getId());
    

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception 
        //处理接收到的消息
        log.info("Received message from client[ID:" + webSocketBeanMap.get(session.getId()).getClientId() +
                "]; Content is [" + message.getPayload() + "].");
        TextMessage textMessage = new TextMessage("Server has received your message.");
        session.sendMessage(textMessage);
    

4、把连接监控类配置进spring，用于程序启动时启动websocket

@Configuration
@EnableWebSocket
public class WebSocketConfiguration implements WebSocketConfigurer 

    @Bean
    public MyWebSocketInterceptor webSocketInterceptor() 
        return new MyWebSocketInterceptor();
    

    @Bean
    public ServerEndpointExporter serverEndpointExporter() 
        return new ServerEndpointExporter();
    

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) 
        webSocketHandlerRegistry
                .addHandler(new MyWebsocketHandler(), "/websocket")
                .addInterceptors(webSocketInterceptor());
    

5、可以做一个简单的鉴权

@Slf4j
public class MyWebSocketInterceptor extends HttpSessionHandshakeInterceptor 

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception 
        logger.info("[MyWebSocketInterceptor#BeforeHandshake] Request from " + request.getRemoteAddress().getHostString());
        if (request instanceof ServletServerHttpRequest) 
            ServletServerHttpRequest serverHttpRequest = (ServletServerHttpRequest) request;
            String token = serverHttpRequest.getServletRequest().getParameter("token");

            //这里做一个简单的鉴权，只有符合条件的鉴权才能握手成功
            if ("123456".equals(token)) 
                // 保存认证用户
                attributes.put("token", token);
                return super.beforeHandshake(request, response, wsHandler, attributes);
             else 
                log.error("token错误。token：[]！", token, new ConsumablesCabinetException("token错误"));
                return false;
            
        
        return super.beforeHandshake(request, response, wsHandler, attributes);
    

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception ex) 
        logger.info("[MyWebSocketInterceptor#afterHandshake] Request from " + request.getRemoteAddress().getHostString());
    

到此，websocket的服务端就完成了，启动工程，就会启动一个websocket服务，现在来实现客户端

客户端

客户端这里也是用java列出，实际上js等各种方式都可以实现
1、先引包

<dependency>
    <groupId>org.java-websocket</groupId>
    <artifactId>Java-WebSocket</artifactId>
    <version>1.5.1</version>
</dependency>

2、客户端连接断开监控类

public class MyWebSocketClient extends WebSocketClient 

    private Logger logger = LoggerFactory.getLogger(getClass());

    public MyWebSocketClient(URI serverUri) 
        super(serverUri);
    

    @Override
    public void onOpen(ServerHandshake serverHandshake) 
        logger.info("[MyWebSocketClient#onOpen]The WebSocket connection is open.");
    

    @Override
    public void onMessage(String s) 
        logger.info("[MyWebSocketClient#onMessage]The client has received the message from server." +
                "The Content is [" + s + "]");
    

    @Override
    public void onClose(int i, String s, boolean b) 
        logger.info("[MyWebSocketClient#onClose]The WebSocket connection is close.");
    

    @Override
    public void onError(Exception e) 
        logger.info("[MyWebSocketClient#onError]The WebSocket connection is error.");
    

3、启动类

public class MyWebSocketMain 

    private static final AtomicInteger count = new AtomicInteger(0);

    public static void main(String[] args) 
        URI uri = URI.create("ws://localhost:8001/websocket?token=123456");
        MyWebSocketClient client = new MyWebSocketClient(uri);
        try 
            // 在连接成功之前会一直阻塞
            client.connectBlocking();

            Timer timer = new Timer();
            MyTimerTask timerTask = new MyTimerTask(client);
            timer.schedule(timerTask, 1000, 2000);
         catch (InterruptedException e) 
            e.printStackTrace();
        
    

    static class MyTimerTask extends TimerTask 

        private final MyWebSocketClient client;

        public MyTimerTask(MyWebSocketClient client) 
            this.client = client;
        

        @Override
        public void run() 
            client.send("Test message from client, the number is " + count.getAndIncrement());
        
    

运行服务端和客户端就可以看到服务器端不停的输出

Received message from client[ID:0]; Content is [Test message from client, the number is 0].

客户端不停的输出

The client has received the message from server.The Content is [Server has received your message.]

这样就说明成功了