devbox
Gausst松鼠会
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

java-websocket服务端、客户端及如何测试_java websocket客户端

作者：Gausst松鼠会 | 2024-03-16 11:20:44

赞

踩

java websocket客户端

JAVA Websocket服务端

1. 导入依赖

<!-- 实际使用包 -->
<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>
<!--websocket作为客户端-->
<dependency>
    <groupId>org.java-websocket</groupId>
    <artifactId>Java-WebSocket</artifactId>
    <version>1.3.5</version>
</dependency>


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

2. 服务端实现

（1）基础版
package com.xie.websocket;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @Description
 * @Date 2022-04-01 11:28
 * @Author xie
 */
@Slf4j
@Component
@ServerEndpoint(value = "/ws/data/{param}")
public class WebSocketServer {

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

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    // 业务service
    private TestService testService = (TestService) ApplicationContextHandle.getBean(TestService.class);

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("param") String param) {
        this.session = session;   //加入set中
        webSocketSet.add(this);
        try {
            while (true) {
                // 处理数据并发送
                List<Map<String, Object>> result = testService.findDataByParam(param);
                sendMessage(JSONObject.toJSONString(result));
                Thread.sleep(3000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到客户端的消息:{}", message);
    }

    /**
     * 发生错误时调用
     */
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }

    /**
     * 发送消息
     */
    public void sendMessage(String message) throws IOException {
        this.session.getAsyncRemote().sendText(message);
    }


    /**
     * 群发自定义消息
     */
    public static void sendInfo(String message) throws IOException {
        for (WebSocketServer item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
                continue;
            }
        }
    }
}



  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
（2）优化版
package com.cdzs.websocket;

import cn.hutool.core.util.RandomUtil;
import com.cdzs.common.core.constant.SecurityConstants;
import com.cdzs.common.core.exception.ServiceException;
import com.cdzs.common.redis.service.CacheService;
import com.cdzs.system.api.RemoteDeviceGrpcService;
import com.cdzs.system.api.domain.device.CdzsDevicesEntity;
import com.cdzs.system.api.domain.grpc.dto.RegisterFaceByAppointDeviceDTO;
import com.cdzs.websocket.util.ByteArraySplitter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
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.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * type: register/verify（注册/人证核验）
 */

@Component
@ServerEndpoint(value = "/person/{type}/{token}")
@Slf4j
public class CdzsPersonWebSocket {

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    private static RemoteDeviceGrpcService remoteDeviceGrpcService;
	//只能通过set方法/getBean()来注入
    @Autowired
    public void setRemoteDeviceGrpcService(RemoteDeviceGrpcService remoteDeviceGrpcService) {
        CdzsPersonWebSocket.remoteDeviceGrpcService = remoteDeviceGrpcService;
    }

    private static CacheService cacheService;

    @Autowired
    public void setCacheService(CacheService cacheService) {
        CdzsPersonWebSocket.cacheService = cacheService;
    }

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

    private static ConcurrentHashMap<String, CdzsPersonWebSocket> socketServersMap = new ConcurrentHashMap<>();

    private String key = "";

    private final static String REGISTER = "register";
    private final static String VERIFY = "verify";

    /**
     * 建立连接
     *
     * @param session
     * @param type    type为register/verify
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("type") String type, @PathParam("token") String token) {
        this.session = session;
        //key = type + ":" + token + ":" + sessionId
        key = type + ":" + token + ":" + session.getId();
        log.info("onOpen->>key:{}", key);
        //判断当前是否已经连接过
        boolean flag = socketServersMap.containsKey(key);
        if (flag) {
            socketServersMap.remove(key);
            socketServersMap.put(key, this);
        } else {
            socketServersMap.put(key, this);
            addOnlineCount(); //在线数加1
        }
        //获取到token，查询该token下的设备
        CdzsDevicesEntity device = cacheService.getCacheObject("choose_device:" + token);
        if (device == null){
            throw new ServiceException("您登录时未选择注册平板");
        }
        String deviceIp = device.getDeviceIp();
        String deviceSn = device.getDeviceSn();
        log.info("token->>deviceIp:{}", deviceIp);
        if (REGISTER.equals(type)) {
            //注册
            RegisterFaceByAppointDeviceDTO registerFaceByAppointDeviceDTO = new RegisterFaceByAppointDeviceDTO();
            registerFaceByAppointDeviceDTO.setDeviceIp(deviceIp);
            registerFaceByAppointDeviceDTO.setSessionId(token);
            registerFaceByAppointDeviceDTO.setIsVerify(false);
            registerFaceByAppointDeviceDTO.setParams(0);
            registerFaceByAppointDeviceDTO.setPersonId("1000" + RandomUtil.randomNumbers(4));
            registerFaceByAppointDeviceDTO.setDeviceSn(deviceSn);
            remoteDeviceGrpcService.regFeaDataByAppointDevice(registerFaceByAppointDeviceDTO, SecurityConstants.INNER);
            log.info("register call success");
        } else if (VERIFY.equals(type)) {
            //人证核验
            remoteDeviceGrpcService.verifyPersonCard(deviceIp, token, SecurityConstants.INNER);
            log.info("verify call success");
        }
        log.info("online number:{}", getOnlineCount());
    }

    /**
     * 主要功能:收到客户端消息后调用的方法
     * 方法名:onMessage
     *
     * @param message 客户端发送过来的消息
     * @param session 返回类型:void
     *                时间:2019年8月14日 上午9:44:42
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("来自客户端的消息->>message:{};sessionId:{}", message, session.getId());
    }

    /**
     * 主要功能:关闭连接
     * 方法名:onClose
     * 返回类型:void
     * 时间:2019年8月14日 上午9:44:58
     */
    @OnClose
    public void onClose() {
        log.info("onClose : {}" + key);
        if (socketServersMap.containsKey(key)) {
            socketServersMap.remove(key);
            subOnlineCount();           //在线数减1
        }
        log.info("有一连接关闭！当前在线连接数为: {}", getOnlineCount());
    }

    /**
     * 发生错误时调用
     * 主要功能:
     * 方法名:onError
     *
     * @param error 返回类型:void
     *              时间:2019年8月14日 上午9:44:07
     */
    @OnError
    public void onError(Throwable error) {
        socketServersMap.remove(key);
        subOnlineCount();
        log.error("webSocket连接发生错误->>errorMessage:{}", error.getMessage());
    }


    /**
     * 人脸注册webSocket
     *
     * @param message   要发送给前端的数据
     * @param sessionId token
     */
    public synchronized void reproductiveSendToRegister(String message, String sessionId) {
        synchronized (this.getClass()) {
            for (Map.Entry<String, CdzsPersonWebSocket> stringMyWebSocketEntry : socketServersMap.entrySet()) {
                try {
                    String key = stringMyWebSocketEntry.getKey();
                    CdzsPersonWebSocket value = stringMyWebSocketEntry.getValue();
                    if (key.contains(sessionId) && key.contains(REGISTER)) {
                        log.info("reproductiveSendToRegister:推送的消息为：" + key);
                        List<byte[]> result = ByteArraySplitter.split(message, 64 * 1024);
                        for (byte[] bytes : result) {
                            String s = new String(bytes);
                            value.session.getBasicRemote().sendText(s);
                        }
                        // 推送结束符
                        value.session.getBasicRemote().sendText("########");
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 人脸注册webSocket
     *
     * @param message   要发送给前端的数据
     * @param sessionId token
     */
    public synchronized void reproductiveSendToVerify(String message, String sessionId) {
        synchronized (this.getClass()) {
            for (Map.Entry<String, CdzsPersonWebSocket> stringMyWebSocketEntry : socketServersMap.entrySet()) {
                try {
                    String key = stringMyWebSocketEntry.getKey();
                    CdzsPersonWebSocket value = stringMyWebSocketEntry.getValue();
                    if (key.contains(sessionId) && key.contains(VERIFY)) {
                        log.info("reproductiveSendToVerify:推送的消息为：" + key);
                        List<byte[]> result = ByteArraySplitter.split(message, 64 * 1024);
                        for (byte[] bytes : result) {
                            String s = new String(bytes);
                            value.session.getBasicRemote().sendText(s);
                        }
                        // 推送结束符
                        value.session.getBasicRemote().sendText("########");
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 获取在线数
     *
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    /**
     * 在线数+1
     */
    public static synchronized void addOnlineCount() {
        CdzsPersonWebSocket.onlineCount++;
    }

    /**
     * 在线数-1
     */
    public static synchronized void subOnlineCount() {
        CdzsPersonWebSocket.onlineCount--;
    }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151
  • 152
  • 153
  • 154
  • 155
  • 156
  • 157
  • 158
  • 159
  • 160
  • 161
  • 162
  • 163
  • 164
  • 165
  • 166
  • 167
  • 168
  • 169
  • 170
  • 171
  • 172
  • 173
  • 174
  • 175
  • 176
  • 177
  • 178
  • 179
  • 180
  • 181
  • 182
  • 183
  • 184
  • 185
  • 186
  • 187
  • 188
  • 189
  • 190
  • 191
  • 192
  • 193
  • 194
  • 195
  • 196
  • 197
  • 198
  • 199
  • 200
  • 201
  • 202
  • 203
  • 204
  • 205
  • 206
  • 207
  • 208
  • 209
  • 210
  • 211
  • 212
  • 213
  • 214
  • 215
  • 216
  • 217
  • 218
  • 219
  • 220
  • 221
  • 222
  • 223
  • 224
  • 225
  • 226
  • 227
  • 228
  • 229
  • 230
  • 231
  • 232
  • 233
  • 对String分片转换为List<byte[]>
package com.cdzs.websocket.util;

import cn.hutool.core.util.ArrayUtil;
import com.cdzs.common.core.utils.StringUtils;
import com.google.common.collect.Lists;

import java.util.List;

public class ByteArraySplitter {
    /**
     * 对String分片转换为List<byte[]>
     * @param source 字符串
     * @param size 分片的长度 单位字节
     * @return
     */
    public static List<byte[]> split(String source, int size) {
        // 存放最终结果
        List<byte[]> result = Lists.newArrayList();

        if (StringUtils.isEmpty(source)) {
            return null;
        }

        byte[] sourceBytes = source.getBytes();
        if (size > sourceBytes.length) {
            result.add(sourceBytes);
            return result;
        }
        // 开始进行split
        int startIndex = 0;
        int endIndex = sourceBytes.length - 1;
        boolean isRunning = true;
        while (isRunning) {
            if ((endIndex + 1) - startIndex > size) {
                result.add(ArrayUtil.sub(sourceBytes, startIndex, startIndex + size));
                startIndex += size;
            } else {
                result.add(ArrayUtil.sub(sourceBytes, startIndex, endIndex + 1));
                isRunning = false;
            }
        }
        return result;
    }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45

3. 客户端实现

package cn.weblab.plugin.socket;

import com.intellij.execution.ui.ConsoleView;
import com.intellij.execution.ui.ConsoleViewContentType;
import com.intellij.openapi.project.Project;
import cn.weblab.plugin.util.ConsoleUtil;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.WebSocket;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.framing.Framedata;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;

public class MyLogWebSocketClient extends WebSocketClient {
    Logger log=LoggerFactory.getLogger(MyLogWebSocketClient.class);
    private Project project;
    private static List<MyLogWebSocketClient> list = new ArrayList<>();

    /**
     * construct a instance
     *
     * @param serverUri need to be connected
     */
    public MyLogWebSocketClient(String serverUri, Project project) throws URISyntaxException {
        super(new URI(serverUri));
        this.project = project;
        this.setConnectionLostTimeout(0);
        if (list.isEmpty()) {
            return;
        }
        for (MyLogWebSocketClient client : list) {
            client.close();
        }
        list.clear();
        list.add(this);
    }

    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        log.info("在线日志socket连接成功");
        ConsoleView logConsole = ConsoleUtil.getLogConsole(project);
        logConsole.print("连接建立成功\n", ConsoleViewContentType.ERROR_OUTPUT);
    }

    @Override
    public void onMessage(String s) {
        s = s.substring(8);
        // if (s.equals("---pong---")) {
        //     return;
        // }
        ConsoleView logConsole = ConsoleUtil.getLogConsole(project);
        logConsole.print(s, ConsoleViewContentType.NORMAL_OUTPUT);
    }

    @Override
    public void onClose(int i, String s, boolean b) {
        log.info("在线日志socket断开");
        ConsoleView logConsole = ConsoleUtil.getLogConsole(project);
        logConsole.print("在线日志socket断开,请重新连接\n", ConsoleViewContentType.ERROR_OUTPUT);
    }

    @Override
    public void onError(Exception e) {
        e.printStackTrace();
    }

    // @Override
    // public void onWebsocketPong(WebSocket conn, Framedata f) {
    //     try {
    //         Thread.sleep(1000 * 10);
    //     } catch (InterruptedException e) {
    //         e.printStackTrace();
    //     }
    //     System.out.println("pong");
    //     this.onWebsocketPing(conn, f);
    //     super.onWebsocketPong(conn, f);
    // }

    @Override
    public void onWebsocketPing(WebSocket conn, Framedata f) {
        try {
            Thread.sleep(1000 * 5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
       send("---pong---");
    }

    @Deprecated
    public static void destroy() {
        if (list.isEmpty())
            return;
        for (MyLogWebSocketClient client : list) {
            client.close();
        }
        list.clear();

    }

    public static void main(String[] args) throws URISyntaxException, InterruptedException {
        // MyWebSocketClient client = new MyWebSocketClient("ws://localhost:8080/websocket/A");
        // boolean b = client.connectBlocking();
        // System.out.println(b);
        // client.send("hello");
        // client.send("hello");
        // client.send("hello");
        // client.send("hello");
        // client.close();
    }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117

4. websocket服务端测试方法

（1）自己编写一个客户端

​ 使用上面的java客户端就可以直接调用测试。

（2）使用postman测试

​ postman版本需要在v8.0以上才有websocket的接口测试

在这里插入图片描述

在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Gausst松鼠会/article/detail/249147
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号