消息服务应用1——java项目使用websocket

news2024/12/30 1:08:56

在当前微服务项目中，由于业务模块众多，消息服务的使用场景变得异常活跃。而WebSocket由于其自身的可靠性强，实时性好，带宽占用更小的优势，在实时通讯应用场景中独占鳌头，加上HTML5标准的普及流行，在前端和后端开发中应用异常广泛。

一.关于WebSocket

1.WebSocket是什么?

WebSocket是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双通信——浏览器和服务器只需要完成一次握手，两者之间就可以创建持久性的连接，并进行双向数据传输。简单来说，它就是html5中的一种新的协议，其实质就是对html的长连接的一种升级。

2.与HTTP比较

HTTP是短连接，WebSocket是长连接；
HTTP通信是单向的，基于请求响应模式，WebSocket支持双向通信；
HTTP和WebSocket底层都是TCP连接。

3.WebSocket优点

websocket连接只需要建立一次，在第一次连接的时候，客户端和服务器会交换必要的信息。
websocket是双向通信的，对比ajax请求还是长轮询，他们都是通过客户端发送请求，服务器响应的形式完成信息的交换，这种模式下服务器处于一种被动的角色。而websocket不存在这个问题，websocket的链接一旦建立，服务器和客户端都可以互推信息。

二. WebSocket使用

文案基于jdk1.8+maven3.8.1+springboot2.7.13而编写

1.引入websocket.jar包

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

2.配置WebSocket

@Configuration
public class MyWebSocketConfig {

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

}

3.WebSocket连接，发送和接收信息

@ServerEndpoint("/websocket/test")
@Component
public class TestWebSocketInfo {

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

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的BigScreenWebSocket对象。
     */
    private static CopyOnWriteArraySet<TestWebSocketInfo> WEB_SOCKET_SET = new CopyOnWriteArraySet<>();
    /**
     * 存储新链接标识。
     */
    public static Map<String, Object> USER_SESSIONS = new HashMap<>(0);
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据。
     */
    private Session session;


    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) throws Exception {
        this.session = session;
        // 生成唯一ID
        String uuid = String.valueOf(UUID.randomUUID());
        // 把唯一标识跟客户端绑定
        USER_SESSIONS.put(uuid, this.session);
        // 加入set中
        WEB_SOCKET_SET.add(this);
        synchronized (this) {
            // 使用加锁的方式保证请求串行计数
            TestWebSocketInfo.ONLINE_COUNT++;
        }
        //需要发送到业务数据
        onMessage(JSONObject.toJSONString("Hello,Websocket"));
        System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        // 从set中删除
        WEB_SOCKET_SET.remove(this);
        boolean removeFlag = WEB_SOCKET_SET.remove(this);
        // 使用加锁的方式保证请求串行计数
        if (removeFlag && BigScreenWebSocketInfo.ONLINE_COUNT > 0) {
            TestWebSocketInfo.ONLINE_COUNT--;
        }
        System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message) {
        sendMessage(message);
    }

    /**
     * 发生错误时调用
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        // 从set中删除
        USER_SESSIONS.remove(session);
        // 从set中删除
        boolean removeFlag = WEB_SOCKET_SET.remove(this);
        USER_SESSIONS.remove(session);
        // 使用加锁的方式保证请求串行计数
        if (removeFlag && BigScreenWebSocketInfo.ONLINE_COUNT > 0) {
            TestWebSocketInfo.ONLINE_COUNT--;
        }
        System.out.println("发生错误");
        error.printStackTrace();
    }


    public void sendMessage(String message) {
        try {
            for (TestWebSocketInfo item : WEB_SOCKET_SET) {
                // 判断链接是否关闭
                if (item.session.isOpen()) {
                    synchronized (item.session) {
                        item.session.getBasicRemote().sendText(message);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

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