SpringBoot整合WebScoket（指定用户接收消息）

什么是WebSocket?

笔者是需要实现指定用户获得实时数据，类似好友邀请助力当前组队情况

spring就有很好的封装，上代码；

引入pom.xml

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

注入bean，WebSocketConfig.java

package com.myelephant.projects.websocket;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * @Author: StephenZhang
 * @date: 2021-01-19 17:54
 */
@Configuration
public class WebSocketConfig {
    /**
     * 给spring容器注入这个ServerEndpointExporter对象
     * 相当于xml：
     * <beans>
     * <bean id="serverEndpointExporter" class="org.springframework.web.socket.server.standard.ServerEndpointExporter"/>
     * </beans>
     * <p>
     * 检测所有带有@serverEndpoint注解的bean并注册他们。
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        System.out.println("我被注入了");
        return new ServerEndpointExporter();
    }
}

自定义一个session绑定对象，相当于指定多个用户接收消息

Clent.java

bagId 组队Id

session 用户会话 唯一

package com.myelephant.projects.websocket;

import com.dandandog.framework.mapstruct.model.MapperVo;
import lombok.Data;
import lombok.EqualsAndHashCode;

import javax.websocket.Session;
import java.io.Serializable;

/**
 * @Author: StephenZhang
 * @date: 2021-01-21 16:01
 */
@Data
@EqualsAndHashCode(callSuper = true)
public class ClientVo extends MapperVo implements Serializable {
    private String bagId;
    private Session session;
}

逻辑实现

目的客服端携带bagId 则生成对象同时拿到对应session

add，如有两个用户同一个bagId则指定用户获取数据即可

通过bagId相等的去发送消息，调用

sendMessage()

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.List;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @Author: StephenZhang
 * @date: 2021-01-19 17:55
 */

@ServerEndpoint(value = "/ws/test/{bagId}")
@Component
public class WebSocketServer {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class);

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

    /**
     * 用线程安全的CopyOnWriteArraySet来存放客户端连接的信息
     */
    private static CopyOnWriteArraySet<ClientVo> socketServers = new CopyOnWriteArraySet<>();

    /**
     * websocket封装的session,信息推送，就是通过它来信息推送
     */
    private Session session;

    /**
     * 服务端的userName,因为用的是set，每个客户端的username必须不一样，否则会被覆盖。
     * 要想完成ui界面聊天的功能，服务端也需要作为客户端来接收后台推送用户发送的信息
     */
    private final static String SYS_USERNAME = "coding";


    /**
     * 用户连接时触发，我们将其添加到
     * 保存客户端连接信息的socketServers中
     *
     * @param session
     * @param bagId
     */
    @OnOpen
    public void open(Session session, @PathParam(value = "bagId") String bagId) {

        this.session = session;
        ClientVo clientVo = new ClientVo();
        clientVo.setBagId(bagId);
        clientVo.setSession(session);
        socketServers.add(clientVo);
        logger.info("客户端:【{}】连接成功", bagId);
        logger.info("有连接成功，当前连接数为：{}", socketServers.size());

    }

    /**
     * 收到客户端发送信息时触发
     * 我们将其推送给客户端(coding)
     * 其实也就是服务端本身，为了达到前端聊天效果才这么做的
     *
     * @param message
     */
    @OnMessage
    public void onMessage(String message) {

        ClientVo client = socketServers.stream().filter(cli -> cli.getSession() == session)
                .collect(Collectors.toList()).get(0);
        sendMessage(client.getBagId() + "<--" + message, SYS_USERNAME);

        logger.info("客户端:【{}】发送信息:{}", client.getBagId(), message);
    }

    /**
     * 连接关闭触发，通过sessionId来移除
     * socketServers中客户端连接信息
     */
    @OnClose
    public void onClose() {
        socketServers.forEach(client -> {
            if (client.getSession().getId().equals(session.getId())) {
                logger.info("客户端:【{}】断开连接", client.getBagId());
                socketServers.remove(client);
                logger.info("有连接关闭，当前连接数为：{}", socketServers.size());
            }
        });
    }

    /**
     * 发生错误时触发
     *
     * @param error
     */
    @OnError
    public void onError(Throwable error) {
        socketServers.forEach(client -> {
            if (client.getSession().getId().equals(session.getId())) {
                socketServers.remove(client);
                logger.error("客户端:【{}】发生异常", client.getBagId());
                error.printStackTrace();
            }
        });
    }

    /**
     * 信息发送的方法，通过客户端的bagId
     * 拿到其对应的session，调用信息推送的方法
     *
     * @param message
     * @param bagId
     */
    public synchronized static void sendMessage(String message, String bagId) {
        socketServers.forEach(client -> {
            if (bagId.equals(client.getBagId())) {
                try {
                    client.getSession().getBasicRemote().sendText(message);
                    logger.info("服务端推送给客户端 :【{}】", client.getBagId(), message);

                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 获取服务端当前客户端的连接数量，
     * 因为服务端本身也作为客户端接受信息，
     * 所以连接总数还要减去服务端
     * 本身的一个连接数
     * <p>
     * 这里运用三元运算符是因为客户端第一次在加载的时候
     * 客户端本身也没有进行连接，-1 就会出现总数为-1的情况，
     * 这里主要就是为了避免出现连接数为-1的情况
     *
     * @return
     */
    public synchronized static int getOnlineNum() {
        return socketServers.stream().filter(client -> !client.getBagId().equals(SYS_USERNAME))
                .collect(Collectors.toList()).size();
    }

    /**
     * 获取在线用户名，前端界面需要用到
     *
     * @return
     */
    public synchronized static List<String> getOnlineUsers() {

        List<String> onlineUsers = socketServers.stream()
                .filter(client -> !client.getBagId().equals(SYS_USERNAME))
                .map(client -> client.getBagId())
                .collect(Collectors.toList());

        return onlineUsers;
    }

    /**
     * 信息群发，我们要排除服务端自己不接收到推送信息
     * 所以我们在发送的时候将服务端排除掉
     *
     * @param message
     */
    public synchronized static void sendAll(String message) {
        //群发，不能发送给服务端自己
        socketServers.stream().filter(cli -> cli.getBagId() != SYS_USERNAME)
                .forEach(client -> {
                    try {
                        client.getSession().getBasicRemote().sendText(message);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });

        logger.info("服务端推送给所有客户端 :【{}】", message);
    }

    /**
     * 多个人发送给指定的几个用户
     *
     * @param message
     * @param persons
     */
    public synchronized static void SendMany(String message, String[] persons) {
        for (String userName : persons) {
            sendMessage(message, userName);
        }
    }
}

贴上前端代码

<html>
<head>
    <meta charset="UTF-8">
    <title>websocket测试</title>
    <script src="http://libs.baidu.com/jquery/2.0.0/jquery.min.js"></script>
    <style type="text/css">
        h3,h4{
            text-align:center;
        }
    </style>
</head>
<body>

<h3>WebSocket测试，客户端接收到的消息如下：</h3>

<textarea id = "messageId" readonly="readonly" cols="150" rows="30" >

</textarea>


<script type="text/javascript">
    var socket;
    if (typeof (WebSocket) == "undefined") {
        console.log("遗憾：您的浏览器不支持WebSocket");
    } else {
        console.log("恭喜：您的浏览器支持WebSocket");
        //实现化WebSocket对象
        //指定要连接的服务器地址与端口建立连接
        //注意ws、wss使用不同的端口。我使用自签名的证书测试，
        //无法使用wss，浏览器打开WebSocket时报错
        //ws对应http、wss对应https。
        if(!socket){
            socket = new WebSocket("ws://localhost:8080/api/ws/test/96325");
        }
        //连接打开事件
        socket.onopen = function() {
            console.log("Socket 已打开");
            socket.send("消息发送测试(From Client)");
        };
        //收到消息事件
        socket.onmessage = function(msg) {
            $("#messageId").append(msg.data+ "
");
            console.log(msg.data  );
            //msg = success则更新 分页获取礼包人数信息and分页获取当前团队人数信息
        };
        //连接关闭事件
        socket.onclose = function() {
            console.log("Socket已关闭");
        };
        //发生了错误事件
        socket.onerror = function() {
            alert("Socket发生了错误");
        }
        //窗口关闭时，关闭连接
        window.unload=function() {
            socket.close();
        };
    }
</script>

</body>
</html>