Spring Boot中使用webSocket实现消息实时推送-摩杜云开发者社区

在写这篇博客之前，我翻过许多资料，读过上的有关于webSocket的许多关于webSocket的相关博文，但是并未找到一个比较完善的，大多数是残缺不堪，讲了相关概念，但是在实际demo过程中，大多数是直接省略的许多代码，然后你直接拿过来，绝大多少是运行不了的，也就是说还得自己琢磨，但是我知道作为一个开发者，虽然这样并不是很麻烦，但是还是比较折磨人的，毕竟大家最希望的事，你提供完整的代码，我直接拿来就用。

这篇博客，就是教你如何使用webSocket完成一对多的通讯，一对一的通讯，以及自己和自己说话，我敢说，如果你学完这篇博客，你根据这篇博客中我提供demo进行开发，可以独立完成一个web在线聊天系统，具体的实现得看你自己怎么去研究了。

什么是webSocket?

WebSocket 是一种在 Web 应用程序中实现双向通信的协议。它允许服务器端和客户端之间建立持久的连接，并通过这个连接进行实时的双向通讯。与传统的 HTTP 请求-响应模式不同，WebSocket 提供了一种更高效、实时的通讯方式。

要实现通过 WebSocket 实现服务器端和客户端的双向通讯，可以遵循以下步骤：

建立 WebSocket 连接：客户端通过 JavaScript 创建一个 WebSocket 对象，并指定服务器的 WebSocket 端点（URL）。客户端通过调用 WebSocket 对象的 onopen 事件监听器来处理 WebSocket 连接成功建立的事件。
发送和接收消息：客户端和服务器端可以通过 WebSocket 对象的 send 方法和 onmessage 事件监听器来发送和接收消息。客户端可以使用 send 方法将消息发送给服务器端，而服务器端可以使用 onmessage 事件监听器来处理接收到的消息。
关闭 WebSocket 连接：客户端和服务器端可以通过调用 WebSocket 对象的 close 方法来主动关闭 WebSocket 连接。另外，WebSocket 连接也可以由服务器端或网络故障等原因自动关闭。

webSocket 案例

导入相关依赖

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


        <dependency>
            <!-- websocket -->
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
        <dependency>
            <!-- fastjson -->
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

配置一个WebSocketConfig

@Configuration
public class WebSocketConfig {
    /**
     * 注入一个ServerEndpointExporter,该Bean会自动注册使用@ServerEndpoint注解申明的websocket endpoint
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

创建一个MyMessage实体类

用来接收相关信息

public class MyMessage {

    private String userId;

    private String message;


    public String getUserId() {
        return userId;
    }

    public void setUserId(String userId) {
        this.userId = userId;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    public MyMessage(String userId, String message) {
        this.userId = userId;
        this.message = message;
    }

    public MyMessage() {
    }

    @Override
    public String toString() {
        return "MyMessage{" +
                "userId='" + userId + '\'' +
                ", message='" + message + '\'' +
                '}';
    }
}

创建一个枚举类WebState

其实使用枚举类的话在这个demo中是多余的，我之所以创建枚举类是用来定义错误码的，也就是说为了看起来更加规范，至于为啥我在一个demo中就定义枚举类对象，这其实是多此一举了，甚至我定义的枚举类参数，我很多都没用到，实际上，也就是让项目看起来更加规范罢了。

public enum WebState {
    /**
     * 连接成功
     */
    CONNECTED("200","连接成功"),
    /**
     * 连接失败
     */
    DISCONNECTED("400","连接失败"),
    /**
     * 连接中
     */
    CONNECTING("199","正在连接"),
    /**
     * 断开连接
     */
    DISCONNECTING("401","连接已断开"),

    /**
     * 未知错误
     */

    UNKNOWN_ERROR("-1","未知错误"),

    /**
     * 成功
     */
    SUCCESS("200","成功"),
    /**
     * 失败
     */
    FAIL("400","失败"),
    /**
     * 参数不合法
     */
    PARAM_ERROR("201","参数不合法"),

    /**
     * 数据库异常
     */
    DATABASE_ERROR("202","数据库异常")


    ;


    /**
     * 错误码
     */
    private String state;
    /**
     * 错误消息
     */
    private String msg;
    WebState(String state, String msg) {
        this.state = state;
        this.msg = msg;
    }

    public String getState() {
        return state;
    }

    public String getMsg() {
        return msg;
    }
}

实现自己对自己说话

/**
 * 前后端交互的类实现消息的接收推送(自己发送给自己)
 *
 * @ServerEndpoint(value = "/test/one") 前端通过此URI和后端交互，建立连接
 */
@Slf4j
@ServerEndpoint(value = "/test/one")
@Component
public class OneWebSocket {

    /**
     * 记录当前在线连接数
     */
    private static AtomicInteger onlineCount = new AtomicInteger(0);

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        onlineCount.incrementAndGet(); // 在线数加1
        log.info("检测到状态码{},对应信息:{},有新连接加入：{}，当前在线人数为：{}",WebState.CONNECTED.getState(),WebState.CONNECTED.getMsg(),session.getId(), onlineCount.get());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        onlineCount.decrementAndGet(); // 在线数减1
        log.info("有一连接关闭：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
        this.sendMessage("Hello, " + message, session);
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误,错误参数为{},错误原因为{}", WebState.UNKNOWN_ERROR.getState(),WebState.UNKNOWN_ERROR.getMsg());
        error.printStackTrace();
    }

    /**
     * 服务端发送消息给客户端
     */
    private void sendMessage(String message, Session toSession) {
        try {
            log.info("服务端给客户端[{}]发送消息{}", toSession.getId(), message);
            toSession.getBasicRemote().sendText(message);
        } catch (Exception e) {
            log.error("服务端发送消息给客户端失败：{}", e);
        }
    }
}

配套自己对自己说话的前端，在Resources目录下新建一个static文件夹，存放index.html以及index.html链接的其他html文件

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>

<body>
<input id="text" type="text" />
<br>

<a href="oneToOne.html">oneToOne</a>
<br>
<a href="oneToMany.html">oneToMany</a>
<br>
<button onclick="send()">Send</button>
<br>
<button onclick="closeWebSocket()">Close</button>
<div id="message"></div>
</body>

<script type="text/javascript">
    var websocket = null;

    //判断当前浏览器是否支持WebSocket, 主要此处要更换为自己的地址
    if ('WebSocket' in window) {
        websocket = new WebSocket("ws://localhost:8021/test/one");
    } else {
        alert('Not support websocket')
    }

    //连接发生错误的回调方法
    websocket.onerror = function() {
        setMessageInnerHTML("error");
    };

    //连接成功建立的回调方法
    websocket.onopen = function(event) {
        //setMessageInnerHTML("open");
    }

    //接收到消息的回调方法
    websocket.onmessage = function(event) {
        setMessageInnerHTML(event.data);
    }

    //连接关闭的回调方法
    websocket.onclose = function() {
        setMessageInnerHTML("close");
    }

    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function() {
        websocket.close();
    }

    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }

    //关闭连接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</script>
</html>

Spring Boot中使用webSocket实现消息实时推送_服务器端

前端测试：

Spring Boot中使用webSocket实现消息实时推送_后端_02

一对一说话

/**
 * 前后端交互的类实现消息的接收推送(自己发送给另一个人)
 *
 * @ServerEndpoint(value = "/test/oneToOne") 前端通过此URI 和后端交互，建立连接
 */
@Slf4j
@ServerEndpoint(value = "/test/oneToOne")
@Component
public class OneToOneWebSocket {

    /** 记录当前在线连接数 */
    private static AtomicInteger onlineCount = new AtomicInteger(0);

    /** 存放所有在线的客户端 */
    private static Map<String, Session> clients = new ConcurrentHashMap<>();

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        onlineCount.incrementAndGet(); // 在线数加1
        clients.put(session.getId(), session);
        log.info("有新连接加入：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        onlineCount.decrementAndGet(); // 在线数减1
        clients.remove(session.getId());
        log.info("有一连接关闭：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message
     *            客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("服务端收到客户端[{}]的消息[{}]", session.getId(), message);
        try {
            //MyMessage myMessage = new MyMessage();
            //myMessage.setUserId(session.getId());
            //myMessage.setMessage(message);
              MyMessage myMessage = JSON.parseObject(message, MyMessage.class);
            if (myMessage != null) {
                Session toSession = clients.get(myMessage.getUserId());
                if (toSession != null) {
                    this.sendMessage(myMessage.getMessage(), toSession);
                }
            }
        } catch (Exception e) {
            log.error("解析失败：{}", e);
        }
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }

    /**
     * 服务端发送消息给客户端
     */
    private void sendMessage(String message, Session toSession) {
        try {
            log.info("服务端给客户端[{}]发送消息[{}]", toSession.getId(), message);
            toSession.getBasicRemote().sendText(message);
        } catch (Exception e) {
            log.error("服务端发送消息给客户端失败：{}", e);
        }
    }

}

对比上边的index.html很容易就可以发现，这个html就仅仅是变化了一个链接，其他的地方都没变化。

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>

<body>
<input id="text" type="text" />
<button onclick="send()">Send</button>
<button onclick="closeWebSocket()">Close</button>
<div id="message"></div>
</body>

<script type="text/javascript">
    var websocket = null;

    //判断当前浏览器是否支持WebSocket, 主要此处要更换为自己的地址
    if ('WebSocket' in window) {
        websocket = new WebSocket("ws://localhost:8021/test/oneToOne");
    } else {
        alert('Not support websocket')
    }

    //连接发生错误的回调方法
    websocket.onerror = function() {
        setMessageInnerHTML("error");
    };

    //连接成功建立的回调方法
    websocket.onopen = function(event) {
        //setMessageInnerHTML("open");
    }

    //接收到消息的回调方法
    websocket.onmessage = function(event) {
        setMessageInnerHTML(event.data);
    }

    //连接关闭的回调方法
    websocket.onclose = function() {
        setMessageInnerHTML("close");
    }

    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function() {
        websocket.close();
    }

    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }

    //关闭连接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</script>
</html>

一对一聊天有个条件，也就是说，我们需要指定我们需要聊天的对象,在这里我们需要传递的值是参数是json的，也就是前端我们在传递的时候需要进行处理一下：
{"message":"你好", "userId":1}，无论你是谁，我都指定只向1号用户发送消息。这样就可以实现一对一发送消息了。

PS： {“message”:“你好”, “userId”:1},很重要，由于是测试，并未实际处理。注意如果在一对一中不以JSON传参，那会报错

Spring Boot中使用webSocket实现消息实时推送_客户端_03

1号用户接收到：

Spring Boot中使用webSocket实现消息实时推送_spring boot_04

Spring Boot中使用webSocket实现消息实时推送_websocket_05

一对多

这个一对多，主要的是多打开几个框进行验证，需要注意的是本demo中，如果是作者发消息，那么作者是收不到消息的，但是只要是通过oneToMany登录的所有用户，他都可以收到消息，除自己本人以外，其他人都可以收到这条消息。

/**
 *
 * 前后端交互的类实现消息的接收推送(自己发送给所有人(不包括自己))
 *
 * @ServerEndpoint(value = "/test/oneToMany") 前端通过此URI 和后端交互，建立连接
 */
@Slf4j
@ServerEndpoint(value = "/test/oneToMany")
@Component
public class OneToManyWebSocket {

    /** 记录当前在线连接数 */
    private static AtomicInteger onlineCount = new AtomicInteger(0);

    /** 存放所有在线的客户端 */
    private static Map<String, Session> clients = new ConcurrentHashMap<>();

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        onlineCount.incrementAndGet(); // 在线数加1
        clients.put(session.getId(), session);
        log.info("有新连接加入：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        onlineCount.decrementAndGet(); // 在线数减1
        clients.remove(session.getId());
        log.info("有一连接关闭：{}，当前在线人数为：{}", session.getId(), onlineCount.get());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message
     *            客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
        this.sendMessage(message, session);
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }

    /**
     * 群发消息
     *
     * @param message
     *            消息内容
     */
    private void sendMessage(String message, Session fromSession) {
        for (Map.Entry<String, Session> sessionEntry : clients.entrySet()) {
            Session toSession = sessionEntry.getValue();
            // 排除掉自己
            if (!fromSession.getId().equals(toSession.getId())) {
                log.info("服务端给客户端[{}]发送消息{}", toSession.getId(), message);
                toSession.getAsyncRemote().sendText(message);
            }
        }
    }
}

将上一个页面的访问后端地址改为： websocket = new WebSocket("ws://localhost:8021/test/oneToMany");

Spring Boot中使用webSocket实现消息实时推送_后端_06

Spring Boot中使用webSocket实现消息实时推送_后端_07

2,4号客户端接受到3号客户端消息：

Spring Boot中使用webSocket实现消息实时推送_spring boot_08