JavaWebsocket-demo

Websocket客户端

pom依赖

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

客户端代码片段


@Component
@Slf4j
public class PositionAlarmListener {
    @Autowired
    private BigScreenService bigScreenService;
    @Autowired
    private ConfigurationSystemService configurationSystemService;

    @Bean
    public WebSocketClient webSocketClient() {
        WebSocketClient wsc = null;
        ThirdPartConfDto thirdPartConfDto = configurationSystemService.getConfig();

        Map<String, String> httpHeaders = new HashMap<>();
        try {
//            String reqUrl = String.format("ws://%s%s?apikey=%s", servicePort, SOCKET_URL, apikey);
            String reqUrl = thirdPartConfDto.getAlarmWebsocketUrl();
            log.info("websocketclient.position.reqUrl:{}",reqUrl);
            wsc = new WebSocketClient(new URI(reqUrl), httpHeaders) {
                @Override
                public void onOpen(ServerHandshake serverHandshake) {
                    log.info("UnmannedPlane==connect==build");
                }

                @Override
                public void onMessage(String message) {
                    log.info("websocketclient.position.receive.message：{}", message);
                    CompletableFuture.runAsync(() -> {
                        try {
                            if (StringUtils.isEmpty(message)) {
                                return;
                            }
//                            JSONObject parse = JSONObject.parseObject(message);
                            ThirdPositionAlarmDto thirdPositionAlarmDto = JSONObject.parseObject(message,ThirdPositionAlarmDto.class);
                            String type = thirdPositionAlarmDto.getType();
                            log.info("websocketclient.position.receive.message-type：{}", type);
                            if (StringUtils.isEmpty(type)) {
                                log.error("websocket.type.is null");
                                return;
                            }
                            if(!type.equals(ThirdPositionAlarmEnum.TYPE_TAG.getCode())){
                                log.error("websocket.type.is not tag");
                                return;
                            }
                           boolean bigScreenPush = bigScreenService.pusdata(thirdPositionAlarmDto,thirdPartConfDto);
                        } catch (Exception e) {
                            log.error("websocketclient.position.error:", e);
                        }
                    });
                }

                @Override
                public void onClose(int i, String s, boolean b) {
                    log.info("websocketclient.position.close code:{} reason:{} {}", i, s, b);
                }

                @Override
                public void onError(Exception e) {
                    log.info("websocketclient.position.connect==error:", e);
                }

            };
            wsc.connect();
            return wsc;
        } catch (Exception e) {
            log.error("websocketclient.position==webSocketClient:", e);
        }

        return wsc;
    }

}

客户端代码片段(新增心跳检测、重连)

客户端代码片段–配置新增

# websocketclient config
#websocket.client.config.wsUrl=ws://10.212.188.45:8880/position
websocket.client.config.wsUrl=ws://10.81.12.100:8090/websocket
websocket.client.config.enableHeartbeat=true
websocket.client.config.heartbeatInterval=20000
websocket.client.config.enableReconnection=true

@Configuration
@ConfigurationProperties(prefix="websocket.client.config")
public class WebsocketClientConfiguration {

   /**
    * websocket server ws://ip:port
    */
   private String wsUrl;
   /**
    * 是否启用心跳监测 默认开启
    */
   private Boolean enableHeartbeat;
   /**
    * 心跳监测间隔 默认20000毫秒
    */
   private Integer heartbeatInterval;
   /**
    * 是否启用重连接 默认启用
    */
   private Boolean enableReconnection;

   public String getWsUrl() {
      return wsUrl;
   }

   public void setWsUrl(String wsUrl) {
      this.wsUrl = wsUrl;
   }

   public Boolean getEnableHeartbeat() {
      return enableHeartbeat;
   }

   public void setEnableHeartbeat(Boolean enableHeartbeat) {
      this.enableHeartbeat = enableHeartbeat;
   }

   public Integer getHeartbeatInterval() {
      return heartbeatInterval;
   }

   public void setHeartbeatInterval(Integer heartbeatInterval) {
      this.heartbeatInterval = heartbeatInterval;
   }

   public Boolean getEnableReconnection() {
      return enableReconnection;
   }

   public void setEnableReconnection(Boolean enableReconnection) {
      this.enableReconnection = enableReconnection;
   }
}

客户端代码片段–客户端创建

@Slf4j
@Configuration
public class WebsocketClientBeanConfig {

    /**
     * 系统配置实现类
     */
    @Autowired
    private ConfigurationSystemService configurationSystemService;

    @Bean
    public WebsocketRunClient websocketRunClient(WebsocketClientConfiguration websocketClientConfiguration){
        String wsUrl = websocketClientConfiguration.getWsUrl();
        Boolean enableHeartbeat = websocketClientConfiguration.getEnableHeartbeat();
        Integer heartbeatInterval = websocketClientConfiguration.getHeartbeatInterval();
        Boolean enableReconnection = websocketClientConfiguration.getEnableReconnection();
        try {
            WebsocketRunClient websocketRunClient = new WebsocketRunClient(new URI(wsUrl));
            websocketRunClient.connect();
            websocketRunClient.setConnectionLostTimeout(0);

            new Thread(()->{
                while (true){
                    try {
                        Thread.sleep(heartbeatInterval);
                        if(enableHeartbeat){
                            websocketRunClient.send("[websocketclient] 心跳检测");
                            log.info("[websocketclient] 心跳检测");
                        }
                    } catch (Exception e) {
                        log.error("[websocketclient] 发生异常{}",e.getMessage());
                        try {
                            if(enableReconnection){
                                log.info("[websocketclient] 重新连接");
                                websocketRunClient.reconnect();
                                websocketRunClient.setConnectionLostTimeout(0);
                            }
                        }catch (Exception ex){
                            log.error("[websocketclient] 重连异常,{}",ex.getMessage());
                        }
                    }
                }
            }).start();

            return websocketRunClient;
        } catch (URISyntaxException ex) {
            log.error("[websocketclient] 连接异常,{}",ex.getMessage());
        }
        return null;
    }


}

客户端代码片段–客户端心跳检测

@Slf4j
@Component
public class WebsocketRunClient extends WebSocketClient {

    /**
     * 大屏推送地址
     */
    @Value("${thirdpart.bigscreen.positionhttpurl}")
    private String httpUrl;
    /**
     * 位置检测距离
     */
    @Value("${thirdpart.positionrange:100}")
    private Double positionrange;
    /**
     * 大屏接口推送实现
     */
    @Autowired
    private BigScreenService bigScreenService;


    public WebsocketRunClient(URI serverUri) {
        super(serverUri);
    }

    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        log.info("[websocketclient] Websocket客户端连接成功");
    }

    @Override
    public void onMessage(String message) {
        log.info("[websocketclient.receive] 收到消息：{}",message);
//        ThirdPartConfDto thirdPartConfDto = configurationSystemService.getConfig();
        ThirdPartConfDto thirdPartConfDto = ThirdPartConfDto.builder()
                .bigScreenHttpUrl(httpUrl)
                .positionRange(positionrange)
                .build();
        CompletableFuture.runAsync(() -> {
            try {
                if (StringUtils.isEmpty(message.trim())) {
                    return;
                }
                if(message.contains("心跳检测")){
                    return;
                }
                List<ThirdPositionAlarmDto> thirdPositionAlarmDtoList = JSONObject.parseArray(message,ThirdPositionAlarmDto.class);
                for(ThirdPositionAlarmDto thirdPositionAlarmDto: thirdPositionAlarmDtoList){
                    String type = thirdPositionAlarmDto.getType();
                    log.info("websocketclient.position.receive.message-type：{}", type);
                    if (StringUtils.isEmpty(type)) {
                        log.error("websocket.type.is null");
                        return;
                    }
                    if(!type.equals(ThirdPositionAlarmEnum.TYPE_TAG.getCode())){
                        log.error("websocket.type.is not tag");
                        return;
                    }
                    boolean bigScreenPush = bigScreenService.pusdata(thirdPositionAlarmDto,thirdPartConfDto);
                }
            } catch (Exception e) {
                log.error("websocketclient.position.error:", e);
            }
        });
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        log.info("[websocketclient] Websocket客户端关闭");
        System.out.println("Connection closed by " + (remote ? "remote peer" : "us") + " Code: " + code + " Reason: " + reason);
    }

    @Override
    public void onError(Exception e) {
        log.info("[websocketclient] Websocket客户端出现异常, 异常原因为：{}",e.getMessage());
    }

Websocket 服务端

服务端pom依赖

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

服务端代码片段，websocket-配置

websocket-服务配置
@Configuration
public class WebSocketConfig {

    /**
     * ServerEndpointExporter注入
     * 该Bean会自动注册使用@ServerEndpoint注解申明的WebSocket endpoint
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

服务端代码片段，websocket-服务端广播消息

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author 谪居马道
 * @describe：websocket,消息广播
 * @date 2025/5/21
 */
@Component
@ServerEndpoint("/websocket")
public class WebSocket {

    private final Logger log = LoggerFactory.getLogger(WebSocket.class);

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<WebSocket> webSocketSet = new CopyOnWriteArraySet<WebSocket>();
    /**         * 连接建立成功调用的方法         */
    @OnOpen
    public void onOpen(Session session){
        this.session=session;
        webSocketSet.add(this);
        //加入set中
        log.info("【WebSocket消息】有新的连接，总数:{}",webSocketSet.size());
    }
    /**         * 连接关闭调用的方法         */
    @OnClose
    public void onClose(){
        webSocketSet.remove(this);
        //从set中删除
        log.info("【WebSocket消息】连接断开,总数:{}",webSocketSet.size());        }
    /**         * 收到客户端消息后调用的方法         * @param message 客户端发送过来的消息         */
    @OnMessage
    public void onMessage(String message ){
        log.info("【WebSocket消息】收到客户端发来的消息:{}",message);
        sendMessage(message);
    }
    public void sendMessage(String message){
        for (WebSocket webSocket:webSocketSet) {
            log.info("【webSocket消息】广播消息,message={}",message);
            try {
                webSocket.session.getBasicRemote ().sendText(message);
            } catch (Exception e) {
                e.printStackTrace ();
            }            }
    }

}

服务端代码片段，websocket-服务端一对一消息

@Component
@ServerEndpoint("/websocket/{terminalId}")
public class WebSocketService {

    private final Logger logger = LoggerFactory.getLogger(WebSocketService.class);

    /**
     * 保存连接信息
     */
    private static final Map<String, Session> CLIENTS = new ConcurrentHashMap<>();
    private static final Map<String, AtomicInteger> TERMINAL_IDS = new HashMap<>();

    /**
     * 需要注入的Service声明为静态，让其属于类
     */
    private static TerminalService terminalService;

    /**
     * 注入的时候，给类的Service注入
     */
    @Autowired
    public void setMchDeviceInfoService(TerminalService terminalService) {
        WebSocketService.terminalService = terminalService;
    }

    @OnOpen
    public void onOpen(@PathParam("terminalId") String terminalId, Session session) throws Exception {
        logger.info(session.getRequestURI().getPath() + "，打开连接开始：" + session.getId());

        //当前连接已存在，关闭
        if (CLIENTS.containsKey(terminalId)) {
            onClose(CLIENTS.get(terminalId));
        }

        TERMINAL_IDS.put(terminalId, new AtomicInteger(0));
        CLIENTS.put(terminalId, session);

        logger.info(session.getRequestURI().getPath() + "，打开连接完成：" + session.getId());

        terminalService.terminal();
    }

    @OnClose
    public void onClose(@PathParam("terminalId") String terminalId, Session session) throws Exception {
        logger.info(session.getRequestURI().getPath() + "，关闭连接开始：" + session.getId());

        CLIENTS.remove(terminalId);
        TERMINAL_IDS.remove(terminalId);

        logger.info(session.getRequestURI().getPath() + "，关闭连接完成：" + session.getId());
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        logger.info("前台发送消息：" + message);

        if ("心跳".equals(message)) {
            //重置当前终端心跳次数
            TERMINAL_IDS.get(message).set(0);
            return;
        }

        sendMessage("收到消息：" + message, session);
    }

    @OnError
    public void onError(Session session, Throwable error) {
        logger.error(error.toString());
    }

    public void onClose(Session session) {
        //判断当前连接是否在线
//        if (!session.isOpen()) {
//            return;
//        }

        try {
            session.close();
        } catch (IOException e) {
            logger.error("金斗云关闭连接异常：" + e);
        }
    }

    public void heartbeat() {
        //检查所有终端心跳次数
        for (String key : TERMINAL_IDS.keySet()) {
            //心跳3次及以上的主动断开
            if ((TERMINAL_IDS.get(key).intValue() >= 3)) {
                logger.info("心跳超时，关闭连接：" + key);
                onClose(CLIENTS.get(key));
            }
        }

        for (String key : CLIENTS.keySet()) {
            //记录当前终端心跳次数
            TERMINAL_IDS.get(key).incrementAndGet();
            sendMessage("心跳", CLIENTS.get(key));
        }
    }

    public void sendMessage(String message, Session session) {
        try {
            session.getAsyncRemote().sendText(message);

            logger.info("推送成功：" + message);
        } catch (Exception e) {
            logger.error("推送异常：" + e);
        }
    }

    public boolean sendMessage(String terminalId, String message) {
        try {
            Session session = CLIENTS.get(terminalId);
            session.getAsyncRemote().sendText(message);

            logger.info("推送成功：" + message);
            return true;
        } catch (Exception e) {
            logger.error("推送异常：" + e);
            return false;
        }
    }
}

Websocket测试工具

postman-测试

参考：
site1： https://maimai.cn/article/detail?fid=1747304025&efid=p7JdUMG2Gi0PrMX7xSXpXw
site2: https://blog.csdn.net/weixin_46768610/article/details/128711019