普通java application中实现websocket客户端

冷不防 2024-04-18 08:26 64阅读 0赞

在项目开发过程中，需要在普通的java application中创建websocket client

依赖的jar包：

<dependency>
                <groupId>org.java-websocket</groupId>
                <artifactId>Java-WebSocket</artifactId>
                <version>1.4.0</version>
            </dependency>

ZKWebSocketClient：

public class ZKWebSocketClient extends WebSocketClient {
        /**
         * 日志
         */
        private static final Logger LOGGER = LoggerFactory.getLogger(ZKWebSocketClient.class);
    
        public ZKWebSocketClient(URI serverUri) {
            super(serverUri);
        }
    
        @Override public void onOpen(ServerHandshake serverHandshake) {
            LOGGER.info("websocket connect open.");
            for (Iterator<String> it = serverHandshake.iterateHttpFields(); it.hasNext(); ) {
                String key = it.next();
                LOGGER.info("{} : {}", key, serverHandshake.getFieldValue(key));
            }
        }
    
        @Override public void onMessage(String message) {
            LOGGER.info("[onMessage] message: {}.", message);
            try {
                FrameTaskQueue.produce(message);
            } catch (InterruptedException e) {
                LOGGER.error("[onMessage] produce msg catch error. cause: ", e);
            }
        }
    
        @Override public void onClose(int i, String s, boolean b) {
            LOGGER.info("websocket connect close.");
        }
    
        @Override public void onError(Exception e) {
            LOGGER.error("websocket error." + e);
        }
    }

程序启动，创建客户端，与websocket服务器进行连接，并每隔30s做一次心跳检测：

/**
         * <websocket客户端初始化>
         *
         * @throws
         */
        public void wsClientInit() {
            LOGGER.info("[LeaderHandler] start to wsClientInit");
            try {
                connectWebsocket();
            } catch (Exception e) {
                LOGGER.error("wsClientInit, error, cause: ", e);
                isOk = false;
            }
    
            // 参数：1、任务体 2、首次执行的延时时间
            //      3、任务执行间隔 4、间隔时间单位
            scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override public void run() {
                    //心跳检测 断线重连
                    heartbeatCheck();
                }
            }, 1, 30, TimeUnit.SECONDS);
    
            LOGGER.info("[LeaderHandler] end to wsClientInit");
        }

/**
         * <连接websocket>
         *
         * @throws
         */
        private void connectWebsocket() {
            try {
                String url = websocketConf.getWebsocketUri()
                    .replaceAll("\\{[a-z,A-Z,0-9]*\\}", UUID.randomUUID().toString().replaceAll("-", ""));
                LOGGER.info("ws url is {}.", url);
                client = new ZKWebSocketClient(new URI(url));
                client.connect();
                while (!client.getReadyState().equals(ReadyState.OPEN)) {
                    LOGGER.info("connecting...");
                    Thread.sleep(1000);
                }
                LOGGER.info("connected.");
                HeartbeatEntity heartbeatEntity = new HeartbeatEntity();
                heartbeatEntity.setPing(IpUtil.getHostAddress());
                client.send(JsonUtil.toJson(heartbeatEntity));
                isOk = true;
            } catch (URISyntaxException e) {
                LOGGER.error("connect websocket catch URISyntaxException. cause: ", e);
                isOk = false;
            } catch (InterruptedException e) {
                LOGGER.error("connect websocket catch InterruptedException. cause: ", e);
                isOk = false;
            }
        }

/**
         * <心跳检测 断线重连>
         *
         * @throws
         */
        private void heartbeatCheck() {
            LOGGER.info("[heartbeatCheck] start to heartbeatCheck");
            if (isOk != null && isOk) {
                try {
                    HeartbeatEntity heartbeatEntity = new HeartbeatEntity();
                    heartbeatEntity.setPing(IpUtil.getHostAddress());
                    client.send(JsonUtil.toJson(heartbeatEntity));
                } catch (Exception e) {
                    LOGGER.error("send msg catch exception. cause: ", e);
                    isOk = false;
                }
            } else {
                // 系统连接失败进行重试
                LOGGER.info("loss connection，reconnect...");
                try {
                    HeartbeatEntity heartbeatEntity = new HeartbeatEntity();
                    heartbeatEntity.setPing(IpUtil.getHostAddress());
                    client.send(JsonUtil.toJson(heartbeatEntity));
                    LOGGER.info("reconnect success.");
                    isOk = true;
                } catch (Exception e) {
                    connectWebsocket();
                    if (isOk != null && isOk) {
                        LOGGER.info("reconnect success.");
                    }
                }
            }
        }