快捷導(dǎo)航

SpringBoot+WebSocket實現(xiàn)即時通訊功能(J2EE方式)

更新時間：2025年01月03日 11:15:38 作者：程序員老馬頭

WebSocket是一種在單個TCP連接上進行全雙工通信的協(xié)議,WebSocket使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加簡單,允許服務(wù)端主動向客戶端推送數(shù)據(jù),本文給大家介紹了SpringBoot+WebSocket實現(xiàn)即時通訊功能(J2EE方式),需要的朋友可以參考下

什么是websocket？

WebSocket是一種在單個TCP連接上進行全雙工通信的協(xié)議。WebSocket通信協(xié)議于2011年被IETF定為標(biāo)準(zhǔn)RFC 6455，并由RFC7936補充規(guī)范。WebSocket API也被W3C定為標(biāo)準(zhǔn)。

WebSocket使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加簡單，允許服務(wù)端主動向客戶端推送數(shù)據(jù)。在WebSocket API中，瀏覽器和服務(wù)器只需要完成一次握手，兩者之間就直接可以創(chuàng)建持久性的連接，并進行雙向數(shù)據(jù)傳輸。

為什么有了HTTP協(xié)議還要WebSocket

HTTP協(xié)議采用的是客戶端（瀏覽器）輪詢的方式，即客戶端發(fā)送請求，服務(wù)端做出響應(yīng)，為了獲取最新的數(shù)據(jù)，需要不斷的輪詢發(fā)出HTTP請求，占用大量帶寬。
WebSocket采用了一些特殊的報頭，使得瀏覽器和服務(wù)器只需要通過“握手”建立一條連接通道后，此鏈接保持活躍狀態(tài)，之后的客戶端和服務(wù)器的通信都使用這個連接，解決了Web實時性的問題，相比于HTTP有一下好處：

一個Web客戶端只建立一個TCP連接
WebSocket服務(wù)端可以主動推送（push）數(shù)據(jù)到Web客戶端
有更加輕量級的頭，減少了數(shù)據(jù)傳輸量

特點

建立在TCP協(xié)議只上，服務(wù)端比較容易實現(xiàn)
于HTTP協(xié)議有良好的兼容性，默認(rèn)端口也是80和443，握手階段使用HTTP協(xié)議，因此握手時不容易屏蔽，能通過各種HTTP代理服務(wù)器
數(shù)據(jù)格式輕量，通信高效且節(jié)省帶寬
支持傳輸文本數(shù)據(jù)和二進制數(shù)據(jù)
沒有同源限制，客戶端可以與任意服務(wù)器通信
也支持加密傳輸，WS+SSL，URL形如wss://

技術(shù)

jdk8
maven
SpringBoot2.6.11
websocket
fastjosn

實現(xiàn)

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.11</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.websocket</groupId>
    <artifactId>springboot_websocket</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot_websocket</name>
    <description>springboot_websocket</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.3</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
 
</project>

websocket核心配置

package com.websocket.springboot_websocket.config;
 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
 
/**
 * @Program: springboot_websocket
 * @ClassName WebSocketConfig
 * @Author: liutao
 * @Description: websocket配置類
 * @Create: 2022-08-19 18:42
 * @Version 1.0
 **/
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

配置websocket服務(wù)

package com.websocket.springboot_websocket.websocket;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
 
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
 
/**
 * @Program: springboot_websocket
 * @ClassName WebsocketServer
 * @Author: liutao
 * @Description: websocket服務(wù)
 * @Create: 2022-08-19 18:52
 * @Version 1.0
 **/
@Slf4j
@Component
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {
//    在線人數(shù)
    private static int onlineCount;
//    當(dāng)前會話
    private Session session;
//    用戶唯一標(biāo)識
    private String userId;
 
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();
 
    /**
     * concurrent包的線程安全set，用來存放每個客戶端對應(yīng)的MyWebSocket對象
     */
    private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap();
 
    /**
     * 為了保存在線用戶信息，在方法中新建一個list存儲一下【實際項目依據(jù)復(fù)雜度，可以存儲到數(shù)據(jù)庫或者緩存】
     */
    private final static List<Session> SESSIONS = Collections.synchronizedList(new ArrayList<>());
 
 /**
  * @methodName: onOpen
  * @description: 建立連接
  * @Author LiuTao
  * @param  [session, userId]
  * @updateTime 2022/8/19 19:31
  * @return void
  * @throws
  **/
 @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        webSocketSet.add(this);
        SESSIONS.add(session);
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            webSocketMap.put(userId,this);
        } else {
            webSocketMap.put(userId,this);
            addOnlineCount();
        }
        log.info("[連接ID:{}] 建立連接, 當(dāng)前連接數(shù):{}", this.userId, getOnlineCount());
    }
 
   /**
    * @methodName: onClose
    * @description: 斷開連接
    * @Author LiuTao
    * @param  []
    * @updateTime 2022/8/19 19:31
    * @return void
    * @throws
    **/
   @OnClose
    public void onClose() {
        webSocketSet.remove(this);
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            subOnlineCount();
        }
        log.info("[連接ID:{}] 斷開連接, 當(dāng)前連接數(shù):{}", userId, getOnlineCount());
    }
 
    /**
     * @methodName: onError
     * @description: 發(fā)送錯誤
     * @Author LiuTao
     * @param  [session, error]
     * @updateTime 2022/8/19 19:32
     * @return void
     * @throws
     **/
    @OnError
    public void onError(Session session, Throwable error) {
        log.info("[連接ID:{}] 錯誤原因:{}", this.userId, error.getMessage());
        error.printStackTrace();
    }
 
   /**
    * @methodName: onMessage
    * @description: 收到消息
    * @Author LiuTao
    * @param  [message]
    * @updateTime 2022/8/19 19:32
    * @return void
    * @throws
    **/
   @OnMessage
    public void onMessage(String message) {
        log.info("[連接ID:{}] 收到消息:{}", this.userId, message);
    }
 
    /**
     * @methodName: sendMessage
     * @description: 發(fā)送消息
     * @Author LiuTao
     * @param  [message, userId]
     * @updateTime 2022/8/19 19:32
     * @return void
     * @throws
     **/
    public void sendMessage(String message,Long userId) {
        WebSocketServer webSocketServer = webSocketMap.get(String.valueOf(userId));
        if (webSocketServer!=null){
            log.info("【websocket消息】推送消息,[toUser]userId={},message={}", userId,message);
            try {
                webSocketServer.session.getBasicRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("[連接ID:{}] 發(fā)送消息失敗, 消息:{}", this.userId, message, e);
            }
        }
    }
 
    /**
     * @methodName: sendMassMessage
     * @description: 群發(fā)消息
     * @Author LiuTao
     * @param  [message]
     * @updateTime 2022/8/19 19:33
     * @return void
     * @throws
     **/
    public void sendMassMessage(String message) {
        try {
            for (Session session : SESSIONS) {
                if (session.isOpen()) {
                    session.getBasicRemote().sendText(message);
                    log.info("[連接ID:{}] 發(fā)送消息:{}",session.getRequestParameterMap().get("userId"),message);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
    /**
     * 獲取當(dāng)前連接數(shù)
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }
 
    /**
     * 當(dāng)前連接數(shù)加一
     */
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }
 
    /**
     * 當(dāng)前連接數(shù)減一
     */
    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
 
}

web接口

package com.websocket.springboot_websocket.web;
 
 
import com.alibaba.fastjson.JSONObject;
import com.websocket.springboot_websocket.websocket.WebSocketServer;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
/**
 * @Program: springboot_websocket
 * @ClassName WebSocketController
 * @Author: liutao
 * @Description: websocket web層
 * @Create: 2022-08-19 19:01
 * @Version 1.0
 **/
@RestController
@RequestMapping("/ws")
public class WebSocketController {
    @Autowired
    private WebSocketServer webSocketServer;
 
    /**
     * 消息發(fā)送
     */
    @GetMapping("/send/{userId}/{msg}")
    public void send(@PathVariable String msg, @PathVariable String userId){
            webSocketServer.sendMessage(JSONObject.toJSONString(msg), Long.valueOf(String.valueOf(userId)));
    }
 
    /**
     * 群發(fā)消息測試(給當(dāng)前連接用戶發(fā)送)
     */
    @GetMapping("/sendMassMessage")
    public void sendMassMessage(){
        WebsocketResponse response = new WebsocketResponse();
        response.setTitle("群發(fā)主題");
        webSocketServer.sendMassMessage(JSONObject.toJSONString(response));
    }
 
    @Data
    @Accessors(chain = true)
    public static class WebsocketResponse {
        private String title;
        private String userId;
        private String userName;
        private int age;
    }
 
}