Java搭建WebSocket的两种方式

时间:2024-04-16 14:51:12

下面分别介绍搭建方法:
一、直接使用Java EE的api进行搭建。
一共3个步骤:
1、添加依赖
<dependency>
    <groupId>javax</groupId>
    <artifactId>javaee-api</artifactId>
    <version>7.0</version>
    <scope>provided</scope>
</dependency>
2、使用注解@ServerEndpoint
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Logger;

/**
* WebSocket连接 sessionKey是url中的参数
*/
@ServerEndpoint("/websocket/{sessionKey}")
public class WebSocket {

    private static final Logger log = Logger.getLogger(WebSocket.class.getName());

    //静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全,用来存放每个客户端对应的MyWebSocket对象。若要实现服务端与单一客户端通信的话,可以使用Map来存放,其中Key可以为用户标识
    private static Map<String, WebSocket> webSockets = new ConcurrentHashMap<>();

    //与某个客户端的连接会话,需要通过它来给客户端发送数据
    private Session session;

    /**
     * 连接建立成功调用的方法
     *
     * @param session    可选的参数。session为与某个客户端的连接会话,需要通过它来给客户端发送数据
     * @param sessionKey url地址参数
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sessionKey") String sessionKey) {

        if (!webSockets.containsKey(sessionKey)) {
            this.session = session;
            webSockets.put(sessionKey, this);
            addOnlineCount();
            log.info("当前websocket连接数:" + onlineCount);
        }
    }

    /**
     * 连接关闭调用的方法
     *
     * @param sessionKey url地址参数
     */
    @OnClose
    public void onClose(@PathParam("sessionKey") String sessionKey) {

        if (webSockets.containsKey(sessionKey)) {
            webSockets.remove(sessionKey);
            subOnlineCount();
            log.info("当前websocket连接数:" + onlineCount);
        }
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     * @param session 可选的参数
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("来自客户端的消息:" + message);
    }

    /**
     * 发生错误时调用
     *
     * @param session 可选的参数
     * @param error   错误消息
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.info("websocket发生错误:" + error);
    }

    /**
     * 该方法没有用注解,是根据自己需要添加的方法。在自己的业务中调用,发送消息给前端。
     *
     * @param sessionKey
     * @param message    返回的结果
     * @throws IOException
     */
    public static void sendMessage(String sessionKey, String message) throws IOException {

        WebSocket webSocket = webSockets.get(sessionKey);

        if (null != webSocket) {

            log.info("websocket发送消息:" + message);

            //同步发送 发送第二条时,必须等第一条发送完成
            webSocket.session.getBasicRemote().sendText(message);

            //异步发送
            //webSocket.session.getAsyncRemote().sendText(message);
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        onlineCount--;
    }

}
3、测试
在线WebSocket测试:http://coolaf.com/tool/chattest
输入地址进行测试即可,举例:ws://localhost:8080/websocket/sessionKey
补充:也可以使用java搭建客户端程序测试,测试的例程放到后面去了~~
二、和springboot整合的WebSocket服务(功能更加强大)
看下效果:<ignore_js_op>
<ignore_js_op>

http://localhost:8080/websocket/sendToUser?username=river&info=你找我干嘛?

 

maven依赖如下:

maven依赖如下:

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-websocket</artifactId>

    <version>2.0.4.RELEASE</version>

</dependency>

1、控制类

package com.boot.river.websocket;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.ResponseBody;

import org.springframework.web.socket.TextMessage;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpSession;

 

@Controller

@RequestMapping(value = "/websocket", method = {RequestMethod.POST, RequestMethod.GET})/*GET请求开放用于测试,最好只允许POST请求*/

public class WebSocketController {

 

    @Autowired

    SpringWebSocketHandler springWebSocketHandler;

 

    /**

     * 登录将username放入session中,然后在拦截器HandshakeInterceptor中取出

     */

    @ResponseBody

    @RequestMapping("/login")

    public String login(HttpServletRequest request, @RequestParam(value = "username") String username, @RequestParam(value = "password") String password) {

        System.out.println("登录:" + username + ":" + password);

        HttpSession session = request.getSession();

        if (null != session) {

            session.setAttribute("SESSION_USERNAME", username);

            return "success";

        } else {

            return "fail";

        }

    }

 

    /**

     * 指定发送

     */

    @ResponseBody

    @RequestMapping("/sendToUser")

    public String send(@RequestParam(value = "username") String username, @RequestParam(value = "info") String info) {

        springWebSocketHandler.sendMessageToUser(username, new TextMessage(info));

        System.out.println("发送至:" + username);

        return "success";

    }

 

    /**

     * 广播

     */

    @ResponseBody

    @RequestMapping("/broadcast")

    public String broadcast(@RequestParam(value = "info") String info) {

        springWebSocketHandler.sendMessageToUsers(new TextMessage("广播消息:" + info));

        System.out.println("广播成功");

        return "success";

    }

}

2、配置类(实现WebSocketConfigurer接口 )

 

@Configuration

@EnableWebSocket

public class SpringWebSocketConfig implements WebSocketConfigurer {

 

    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {

        registry.addHandler(getSpringWebSocketHandler(), "/websocket/server")

                .addInterceptors(getInterceptor()).setAllowedOrigins("*");

 

        registry.addHandler(getSpringWebSocketHandler(), "/sockjs/server").setAllowedOrigins("*")

                .addInterceptors(getInterceptor()).withSockJS();

    }

 

    @Bean

    public SpringWebSocketHandler getSpringWebSocketHandler() {

        return new SpringWebSocketHandler();

    }

 

    @Bean

    public SpringWebSocketHandlerInterceptor getInterceptor() {

        return new SpringWebSocketHandlerInterceptor();

    }

}

3、处理类(实现了WebSocketHandler接口)

package com.boot.river.websocket;

import org.springframework.web.socket.CloseStatus;

import org.springframework.web.socket.TextMessage;

import org.springframework.web.socket.WebSocketSession;

import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;

import java.util.HashMap;

import java.util.Map;

public class SpringWebSocketHandler extends TextWebSocketHandler {

    /**

     * 存储用户id和其对应的session

     */

    private static final Map<String, WebSocketSession> users = new HashMap<>();

 

    /**

     * 用户名key值

     */

    private static final String USER_ID = "WEBSOCKET_USERID";

 

    /**

     * 连接建立后触发

     */

    @Override

    public void afterConnectionEstablished(WebSocketSession session) {

        System.out.println("成功建立websocket连接!");

        String userId = (String) session.getAttributes().get(USER_ID);//取出在拦截器中存储的username

        users.put(userId, session);

        System.out.println("当前线上用户数量:" + users.size());

    }

 

    /**

     * 关闭连接时触发

     */

    @Override

    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) {

        String userId = (String) session.getAttributes().get(USER_ID);

        System.out.println("用户" + userId + "已退出!");

        users.remove(userId);

        System.out.println("剩余在线用户" + users.size());

    }

 

    /**

     * 接收消息

     */

    @Override

    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {

        super.handleTextMessage(session, message);

        System.out.println("收到消息:" + message);

        if (message.getPayload().contains("在吗")) {

           session.sendMessage(new TextMessage("对方不在线!"));

        }

    }

    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {

        if (session.isOpen()) {

            session.close();

        }

        System.out.println("传输出现异常,关闭websocket连接... ");

        String userId = (String) session.getAttributes().get(USER_ID);

        users.remove(userId);

    }

    public boolean supportsPartialMessages() {

        return false;

    }

    /**

     * 给某个用户发送消息

     */

    public void sendMessageToUser(String userId, TextMessage message) {

        for (String id : users.keySet()) {

            if (id.equals(userId)) {

                try {

                    if (users.get(id).isOpen()) {

                        users.get(id).sendMessage(message);

                    }

                } catch (IOException e) {

                    e.printStackTrace();

                }

                break;

            }

        }

    }

 

    /**

     * 给所有在线用户发送消息

     */

    public void sendMessageToUsers(TextMessage message) {

        for (String userId : users.keySet()) {

            try {

                if (users.get(userId).isOpen()) {

                    users.get(userId).sendMessage(message);

                }

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

    }

}

4、拦截器(实现HandshakeInterceptor接口)

package com.boot.river.websocket;

import org.springframework.http.server.ServerHttpRequest;

import org.springframework.http.server.ServerHttpResponse;

import org.springframework.http.server.ServletServerHttpRequest;

import org.springframework.web.socket.WebSocketHandler;

import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;

 

import javax.servlet.http.HttpSession;

import java.util.Map;

 

public class SpringWebSocketHandlerInterceptor extends HttpSessionHandshakeInterceptor {

    @Override

    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,

                                   Map<String, Object> attributes) throws Exception {

        System.out.println("Before Handshake");

        if (request instanceof ServletServerHttpRequest) {

            ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;

            HttpSession session = servletRequest.getServletRequest().getSession(false);//获取session时,如果没有则返回null

            if (session != null) {

                String userName = (String) session.getAttribute("SESSION_USERNAME");//在登录时保存的用户名

                if (userName != null) {

                    attributes.put("WEBSOCKET_USERID", userName);//放入attributes中,可以在处理器的WebSocketSession中取出

                }

            }

        }

        return super.beforeHandshake(request, response, wsHandler, attributes);

    }

 

    @Override

    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,

                               Exception ex) {

        super.afterHandshake(request, response, wsHandler, ex);

        System.out.println("after Handshake");

    }

}

参考:

https://my.oschina.net/u/3445245/blog/3003208

https://blog.****.net/runbat/article/details/80985944

https://docs.spring.io/spring/docs/5.0.0.BUILD-SNAPSHOT/spring-framework-reference/html/websocket.html

下面分别介绍websocket的java客户端请求和js客户端请求

1、java客户端

添加依赖:

<dependency>

      <groupId>org.java-websocket</groupId>

      <artifactId>Java-WebSocket</artifactId>

      <version>1.4.0</version>

</dependency>

package com.river.websocket;

import org.java_websocket.enums.ReadyState;

import java.net.URISyntaxException;

/**

* @author river

* @date 2019-12-6

*/

public class Client {

    public static void main(String[] args) throws URISyntaxException, InterruptedException {

        MyWebSocketClient client = new MyWebSocketClient("ws://localhost:8080/websocket/server");

        client.connect();

        while (client.getReadyState() != ReadyState.OPEN) {

            System.out.println("连接状态:" + client.getReadyState());

            Thread.sleep(100);

        }

        client.send("测试数据!");

        client.close();

    }

}

继承WebsocketClient

package com.river.websocket;

import org.java_websocket.client.WebSocketClient;

import org.java_websocket.handshake.ServerHandshake;

import java.net.URI;

import java.net.URISyntaxException;

public class MyWebSocketClient extends WebSocketClient {

    MyWebSocketClient(String url) throws URISyntaxException {

        super(new URI(url));

    }

    @Override

    public void onOpen(ServerHandshake shake) {

        System.out.println(shake.getHttpStatusMessage());

    }

    @Override

    public void onMessage(String paramString) {

        System.out.println(paramString);

    }

 

    @Override

    public void onClose(int paramInt, String paramString, boolean paramBoolean) {

        System.out.println("关闭");

    }

 

    @Override

    public void onError(Exception e) {

        System.out.println("发生错误");

    }

}

2、js客户端

 

<!DOCTYPE html>

<html>

<head>

    <meta http-equiv="Content-Type" content="text/html; charset=UTF-8"/>

    <title>websocket</title>

    <script type="text/javascript" src="http://cdn.bootcss.com/jquery/3.1.0/jquery.min.js"></script>

    <script type="text/javascript" src="http://cdn.bootcss.com/sockjs-client/1.1.1/sockjs.js"></script>

    <script type="text/javascript">

        var websocket = null;

        if (\'WebSocket\' in window) {

            websocket = new WebSocket("ws://localhost:8080/websocket/server");

        } else if (\'MozWebSocket\' in window) {

            websocket = new MozWebSocket("ws://localhost:8080/websocket/server");

        } else {

            websocket = new SockJS("http://localhost:8080/sockjs/server");

        }

        websocket.onopen = onOpen;

        websocket.onmessage = onMessage;

        websocket.onerror = onError;

        websocket.onclose = onClose;

 

        function onOpen(event) {

            alert(event.type);

        }

 

        function onMessage(messageEvent) {

            alert(messageEvent.data);

        }

 

        function onError(event) {

        }

 

        function onClose(closeEvent) {

            alert(closeEvent.reason);

        }

 

        function doSendUser() {

            if (websocket.readyState === websocket.OPEN) {

                var msg = document.getElementById("inputMsg").value;

                websocket.send(msg);//发送消息

                alert("发送成功!");

            } else {

                alert("连接失败!");

            }

        }

        window.close = function () {

            websocket.onclose();

        };

        function websocketClose() {

            websocket.close();

            alert("连接关闭");

        }

    </script>

</head>

<body>

请输入:<input id="inputMsg" name="inputMsg"/>

<button οnclick="doSendUser();">发送</button>

<button οnclick="websocketClose();">关闭连接</button>

</body>

</html>

补充:登录方式除了在Controller中处理,还可以在Websocket中接收到的消息进行登录处理。

更多java学习资料可关注:itheimaGZ获取