[网络协议]WebSocket的实践

直接上代码了

依赖：

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

当前code的Package 在所展示的code中 已经剔除。

import cn.hutool.core.util.ObjectUtil;
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.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author wzq
 * @event
 * @describe
 * @Date on 2022/3/1 13:58
 */
@ServerEndpoint(value = "/webSocket/{userCode}")
@Component
public class MainWebSocket {
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
    * */
    private Session session;

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    private String currentUser;

    /**
     * @Author: wzq
     * @Description: 连接建立成功调用的方法
     * @Date: 2022-3-1 14:00:53
     * @Param: [session]
     * @return: void
     **/
    @OnOpen
    public void onOpen(@PathParam("userCode") String userCode, Session session) {
        if(ObjectUtil.isNull(userCode)){
            userCode = String.valueOf(0);
        }
        this.currentUser = userCode;
        this.session = session;
        WebSocketUtil.addSocket(this);  //加入到set中
        log.info("有新连接{}加入！当前{}人在线,新用户userCode:{}加入,", session.getId(), WebSocketUtil.getOnlineCount(),userCode);
        log.info("此连接时间:{}！", getNow());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        WebSocketUtil.removeSocket(this);  //从set中删除
        log.info("有一连接关闭！当前还有{}人在线", WebSocketUtil.getOnlineCount());
    }

    /**
     * @Author: wzq
     * @Description: 收到客户端消息后调用的方法
     * @Date: 2022-3-1 14:00:45
     * @Param: [message, sessionCurrent]
     * @return: void
     **/
    @OnMessage
    public void onMessage(String message, Session sessionCurrent) {
        log.info("来自客户端{}的消息:{}", sessionCurrent.getId(), message);
        /**
         * 向客户端群发消息
         */
        for (MainWebSocket item : WebSocketUtil.getWebSocketSet()) {
            try {
                //向发件人单独回应
                if (item.session.getId().equals(sessionCurrent.getId())) {
                    item.sendMessage("我是服务器，我已经收到你的信息---" + message);
                    return;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * @Author: wzq
     * @Description: 发生错误时调用
     * @Date: 2022-3-1 14:00:34
     * @Param: [session, error]
     * @return: void
     **/
    @OnError
    public void onError(Session session, Throwable error) {
        log.info("关闭{}发生错误", session.getId());
        error.printStackTrace();
    }


    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
        //this.session.getAsyncRemote().sendText(message);
    }

    public void sendMessageToUser(String message,String userCode) throws IOException {
        for (MainWebSocket item : WebSocketUtil.getWebSocketSet()) {
            System.out.println("----->code:"+item.currentUser);
            if(item.currentUser.equals(userCode)){
                item.session.getBasicRemote().sendText(message);
            }
        }
    }

    public String getNow() {
        Date date = new Date();
        DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String time = format.format(date);
        return time;
    }
}

/**
 * @author wzq
 * @event
 * @describe
 * @Date on 2022/3/1 14:02
 */
public interface SendWebSocket {
    /**
     * @Author: wzq
     * @Description: 群发自定义消息, 向当前所有在线的用户发信息
     * @Date: 2022-3-1 14:02:54
     * @Param: [message]
     * @return: void
     **/
    void sendToAll(String message);

    /**
     * 给指定的人发送消息
     * @param message String
     * @param userCode String
     */
    void sendToUser(String message,String userCode);

    /**
     * s2022-3-2 11:13:02
    * */
    String getNow();

}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author wzq
 * @event
 * @describe
 * @Date on 2022/3/1 14:03
 */
@Component
public class SendWebSocketImpl implements SendWebSocket{

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    @Override
    public void sendToAll(String message) {
        log.info("当前{}人在线", WebSocketUtil.getOnlineCount());
        for (MainWebSocket item : WebSocketUtil.getWebSocketSet()) {
            try {
                item.sendMessage(message);
                System.out.println("sendToAll 日志:-message为:"+message);
            } catch (IOException e) {
                continue;
            }
        }
    }

    @Override
    public void sendToUser(String message,String userCode) {
        for (MainWebSocket item : WebSocketUtil.getWebSocketSet()) {
            try {
                item.sendMessageToUser(message,userCode);
                System.out.println("sendToUser 日志:---userCode:"+userCode+"message为:"+message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public String getNow() {
        Date date = new Date();
        DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String time = format.format(date);
        return time;
    }

}

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

/**
 * @author wzq
 * @event  bean实例，让容器自动加载，也可以放在Applition的启动容器下
 * @describe
 * @Date on 2022/3/1 13:55
 */
@Configuration
public class WebSocketConfig {

    /**
     * 会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
     * 要注意，如果使用独立的servlet容器，
     * 而不是直接使用springboot的内置容器，
     * 就不要注入ServerEndpointExporter，因为它将由容器自己提供和管理。
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author wzq
 * @event
 * @describe
 * @Date on 2022/3/1 11:30
 */
public class WebSocketUtil {

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

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
    * */
    private static CopyOnWriteArraySet<MainWebSocket> webSocketSet = new CopyOnWriteArraySet<>();

    /**
     * 获取所有WebSocket对象
     * */
    public static CopyOnWriteArraySet<MainWebSocket> getWebSocketSet() {
        return webSocketSet;
    }

    /**
     * 添加一个WebSocket对象
     * */
    public static void addSocket(MainWebSocket webSocket) {
        webSocketSet.add(webSocket);
        addOnlineCount();//在线数加1
    }

    /**
     * 移除一个WebSocket对象
     * */
    public static void removeSocket(MainWebSocket webSocket) {
        webSocketSet.remove(webSocket);
        //在线数减1
        subOnlineCount();
    }

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

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

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

测试：

import com.xacf.athena.science.core.socket.SendWebSocket;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


/**
 * @author wzq
 * @event    WebSocket通讯
 * @describe
 * @Date on 2022/3/1 11:36
 */

@Api(value = "WebSocket通讯", tags = {"WebSocket通讯"})
@RestController
@RequestMapping("/webSocket")
public class WebSocketController {

    @Autowired
    private SendWebSocket sendWebSocket;

    @ApiOperation("群发消息")
    @PostMapping("/sendMass")
    public void sendMass(String msg) {
        sendWebSocket.sendToAll(msg);
    }

    @ApiOperation("发消息给指定用户")
    @PostMapping("/sendUser")
    public void sendUser(String msg,String userCode) {
        sendWebSocket.sendToUser(msg,userCode);
    }
}