1.背景
Tio的官网
https://www.tiocloud.com/tio/index.html
由于之前做了一个停车缴费的系统，停车场终端的设备和控制开闸的程序是海康的，公司想节省那点钱要自己搞一个停车缴费的替换海康的那个平台，在这个项目中最主要的一个环节就是对车场的车道终端的开关等指令如何下达且让终端正确执行？所以我们自己写了一个java应用程序部署在车场终端的电脑上，技术选型采用半开源的Tio，采用websocket协议通过客户端向服务端发送心跳来维持客户端的在线，服务端存在一个心跳检测的线程每隔2秒检测一次最后收到客户端心跳包的时间是否超过心跳检测时长，如果超过心跳检测超时时长则会把客户端踢下线，举个栗子：当前时间-最后一次收到心跳包的时间>心跳检测时长（默认是2分钟，该参数可以灵活配置），那么改客户端就会被强制的下线，这个心跳检测的tio的服务端依赖的源码里面有相关的代码，有兴趣的可以去阅读下Tio的源码；服务端可以通过给在线的客户端发送一些指令消息给客户端然后客户端在本地通过httpUtils直接调用海康的终端设备的api接口操作车道的开关锁等一系列的操作，发送给客户端的消息采用AES加密防止被攻击和消息明文传输，基本的原理大概就是这种，在该项目中如果客户端掉线了服务端感知不到，用户支付了停车费后下发的消息就会失败，车场车道出口的杆就不会抬起，所以就需要一个客户端掉线预警的功能通知到技术人员如果有某个终端掉线了10分钟内收到10次相同的告警消息就需要去排查车场的终端程序是否掉线或者网络出现抖动，重启车场终端应用程序后确保客户端能重新上线，不会影响用户支付抬杆出去。
Tio的底层是基于Aio的异步io实现了websocket协议，虽然说学习难度没有netty那么的难，但是这种半开源半商业化的产品开源出来的东西真的是有很大的坑在里面的，用一个简单的词汇来形容“阉割版”，就比如：
1.服务端没有知道客户端是否在线还是没有在线的功能？如果服务端是单节点部署可以修改下服务端的源码获取到Users对象轻松知道客户端是否在线，但是如果服务端是多节点部署，那么客户端是会在服务端的一个节点上上线和超时后被踢下线然后另一个节点又上线，所以服务端多节点单节点方式修改源码的方式就搞不了。
2.服务端的消息是采用的是redis的发布订阅使用的是一个订阅的topic主题，多个业务系统都使用一个消息主题不太好？修改源码可以配置topic避免各个业务系统的消息相互影响。
3.提供的客户端的依赖中有心跳、重试的代码，但是基本上是用不成的，我是修改调试过它的源码的，会出一些莫名其妙的错误，服务端和客户端的注释需要使用者自行实现心跳和重试机制？
以上这几点是Tio存在的问题，有问题那只能给使用者来填坑了，决绝不了就只能重新技术选型了，没有技术选的只能自己造了，那难度和时间成本就大了。
2.服务端实现
服务端采用的技术有：
服务端demo工程目录结构如下：
大致思路如下：客户端上线在服务端完成握手逻辑时候将客户端的信息保存在redis中，过期时间设置为两分钟，客户端的心跳间隔时长是2s，xxl-job任务每隔1分钟扫描一次redis中指定key中是否包含已经接入的车场id配置，如果redis中没有指定的车场的id说明车场的程序已经下线，此时需要发送消息到钉钉群中告警，问题来了，如果某个客户端一直掉线，那不是每隔一分钟钉钉群里都会收到告警消息，这就会造成了钉钉群的消息轰炸，所以需要一个扫描掉线的发消息的次数限制，掉线次数超过10就不会在向钉钉群发送告警消息，当客户端重新上线握手的时候把redis中的掉线扫描次数清空；该功能会有一定的误判产生，扫描任务刚好扫的时候，客户端刚好在服务端的一个节点下线然后在服务端的另外一个节点上线这种情况就还是会发送几次钉钉告警的消息，这种是避免不了的，可以归结为正常的情况忽略钉钉告警消息，不正常的是10分钟内收到同一个终端的10告警消息这种就是终端确实是掉线的告警了。
2.1 服务端pom依赖和yml配置
pom依赖
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.0modelVersion>
<parent>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-parentartifactId>
<version>2.3.2.RELEASEversion>
<relativePath/>
parent>
<groupId>com.dytz.barrier.gate.servergroupId>
<artifactId>barrier-gate-serverartifactId>
<version>0.0.1-SNAPSHOTversion>
<name>barrier-gate-servername>
<packaging>jarpackaging>
<properties>
<java.version>1.8java.version>
<java.version>1.8java.version>
<spring-cloud.version>Hoxton.SR6spring-cloud.version>
<mybatis-plus-generator.version>3.3.2mybatis-plus-generator.version>
<spring-cloud-alibaba.version>2.2.1.RELEASEspring-cloud-alibaba.version>
<commons-collections4.vsersion>4.1commons-collections4.vsersion>
properties>
<repositories>
<repository>
<id>nexusid>
<url>xxxxxxxurl>
<releases>
<enabled>trueenabled>
<updatePolicy>alwaysupdatePolicy>
releases>
<snapshots>
<enabled>trueenabled>
<updatePolicy>alwaysupdatePolicy>
snapshots>
repository>
repositories>
<distributionManagement>
<repository>
<id>nexus-snapshotsid>
<name>Nexus snapshotsname>
<url>xxxxxxxxurl>
repository>
<snapshotRepository>
<id>nexus-snapshotsid>
<url>xxxxxxxxurl>
snapshotRepository>
distributionManagement>
<profiles>
<profile>
<id>uatid>
<properties>
<env>uatenv>
properties>
<activation>
<activeByDefault>trueactiveByDefault>
activation>
profile>
<profile>
<id>prodid>
<properties>
<env>prodenv>
properties>
profile>
profiles>
<dependencies>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-webartifactId>
dependency>
<dependency>
<groupId>org.projectlombokgroupId>
<artifactId>lombokartifactId>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-testartifactId>
<scope>testscope>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starterartifactId>
dependency>
<dependency>
<groupId>com.alibabagroupId>
<artifactId>fastjsonartifactId>
<version>1.2.83_noneautotypeversion>
dependency>
<dependency>
<groupId>cn.hutoolgroupId>
<artifactId>hutool-allartifactId>
<version>5.8.4version>
dependency>
<dependency>
<groupId>org.projectlombokgroupId>
<artifactId>lombokartifactId>
dependency>
<dependency>
<groupId>org.t-iogroupId>
<artifactId>tio-websocket-spring-boot-starterartifactId>
<version>3.6.0.v20200315-RELEASEversion>
dependency>
<dependency>
<groupId>org.apache.commonsgroupId>
<artifactId>commons-collections4artifactId>
<version>4.1version>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-testartifactId>
dependency>
<dependency>
<groupId>org.testnggroupId>
<artifactId>testngartifactId>
<version>RELEASEversion>
<scope>compilescope>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-testartifactId>
dependency>
<dependency>
<groupId>com.xuxueligroupId>
<artifactId>xxl-job-coreartifactId>
<version>2.2.0version>
dependency>
<dependency>
<groupId>redis.clientsgroupId>
<artifactId>jedisartifactId>
<version>3.3.0version>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-data-redisartifactId>
dependency>
<dependency>
<groupId>org.springframework.cloudgroupId>
<artifactId>spring-cloud-starter-configartifactId>
<version>3.1.5version>
dependency>
<dependency>
<groupId>org.apache.commonsgroupId>
<artifactId>commons-collections4artifactId>
<version>${commons-collections4.vsersion}version>
dependency>
<dependency>
<groupId>com.alibaba.cloudgroupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discoveryartifactId>
dependency>
<dependency>
<groupId>com.alibaba.cloudgroupId>
<artifactId>spring-cloud-starter-alibaba-nacos-configartifactId>
dependency>
<dependency>
<groupId>org.springframework.cloudgroupId>
<artifactId>spring-cloud-contextartifactId>
<version>2.2.3.RELEASEversion>
dependency>
<dependency>
<groupId>org.springframework.cloudgroupId>
<artifactId>spring-cloud-starter-bootstrapartifactId>
<version>3.0.1version>
dependency>
dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloudgroupId>
<artifactId>spring-cloud-dependenciesartifactId>
<version>${spring-cloud.version}version>
<type>pomtype>
<scope>importscope>
dependency>
<dependency>
<groupId>com.alibaba.cloudgroupId>
<artifactId>spring-cloud-alibaba-dependenciesartifactId>
<version>${spring-cloud-alibaba.version}version>
<type>pomtype>
<scope>importscope>
dependency>
dependencies>
dependencyManagement>
<build>
<resources>
<resource>
<directory>src/main/resourcesdirectory>
<excludes>
<exclude>application-*.ymlexclude>
excludes>
resource>
<resource>
<directory>src/main/resourcesdirectory>
<filtering>truefiltering>
<includes>
<include>application.ymlinclude>
<include>application-${env}*.ymlinclude>
includes>
resource>
resources>
<plugins>
<plugin>
<groupId>org.apache.maven.pluginsgroupId>
<artifactId>maven-deploy-pluginartifactId>
<version>2.8.2version>
<configuration>
<skip>trueskip>
configuration>
plugin>
<plugin>
<groupId>org.apache.maven.pluginsgroupId>
<artifactId>maven-surefire-pluginartifactId>
<version>2.22.2version>
<configuration>
<skipTests>trueskipTests>
configuration>
plugin>
<plugin>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-maven-pluginartifactId>
<version>2.3.7.RELEASEversion>
<executions>
<execution>
<goals>
<goal>repackagegoal>
goals>
execution>
executions>
plugin>
plugins>
build>
project>
yml配置
fastjson:
parser:
safeMode: true
# 日记配置
logging:
level:
com.dytz.barrier.gate.server: debug
spring:
main:
allow-circular-references: true
application:
name: barrier-gate-server2
port: ${SERVER_PORT:8080}
cloud:
nacos:
discovery:
enabled: true
server-addr: ${cloud.nacos_url}
service: ${spring.application.name}
namespace: ${cloud.nacos_namespace}
config:
server-addr: ${cloud.nacos_url}
file-extension: yaml
namespace: ${cloud.nacos_namespace}
redis:
database: 6
host: #redis地址
port: 6389
password: #redis密码
jedis:
pool:
max-active: 200
max-idle: 20
max-wait: 2000
min-idle: 5
#Tio配置
tio:
websocket:
server:
port: 19009
heartbeat-timeout: 20000
cluster:
enabled: true
redis:
ip: #redis地址
port: 6389
pool-size: 5
minimum-idle-size: 2
password: #redis密码
user: true
group: true
xxl:
job:
admin:
addresses: #xxl-job的服务端地址
executor:
appname: barrier-gate-server2 #xxl-job的执行器应用名称
port: 10031 #xxl-job的执行器端口
logpath: /logs/xxl
logretentiondays: 5
dingding:
accessToekn: # 钉钉群的机器人的token
baseUrl: https://oapi.dingtalk.com/robot/send
secret: # 钉钉群的机器人的秘钥
cloud:
nacos_url: #nacos地址
nacos_namespace: #nacos的namespace的id
2.2 tio服务端WsMsgHandlerServer
服务端需要在启动主类上加上：@EnableTioWebSocketServer注解，开启Tio的服务端功能
package com.dytz.barrier.gate.server.ws.handler;
import com.alibaba.fastjson.JSON;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.http.common.HttpRequest;
import org.tio.http.common.HttpResponse;
import org.tio.websocket.common.WsRequest;
import org.tio.websocket.server.handler.IWsMsgHandler;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
@Slf4j
@Component
public class WsMsgHandlerServer implements IWsMsgHandler {
@Autowired
private RedisTemplate redisSSTemplate;
@Autowired
private RedisTemplate redisTemplate;
private final String CLINET_KEY = "PARK:";
private final String CLINET_ERROR_COUNT_KEY = "PARK:ERROR:COUNT:";
/**
* ws 会话握手校验，权限登录检查，检查通过后会话绑定userId
*
* @param httpRequest
* @param httpResponse
* @param channelContext
* @return
* @throws Exception
*/
@Override
public HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
String clientip = httpRequest.getClientIp();
log.info(">>>>>>>>>> clientip : {}", clientip);
log.info(">>>>>>>>>> getHeaders : {}", httpRequest.getHeaders());
String parkId = httpRequest.getHeaders().get("parkid");
System.out.println(parkId);
if (Objects.nonNull(parkId)) {
log.info("ws request parkId : {} ", parkId);
String useridType = "PARK_" + parkId;
//Tio.removeUser(channelContext.tioConfig, useridType, "重复登录，如果上次会话没断开主动下线上次会话！");
channelContext.setUserid(useridType);
Tio.bindUser(channelContext, useridType);
log.debug("收到来自{}的ws握手包\r\nhttpRequest{}", clientip, httpRequest);
if (redisSSTemplate.hasKey(CLINET_KEY + parkId)) {
//String clientId = String.valueOf(redisSSTemplate.opsForValue().get(CLINET_KEY + parkId));
//先删除
/*RedisOperations ops1 = redisSSTemplate.boundHashOps(CLINET_KEY + parkId).getOperations();
ValueOperations sv1 = ops1.opsForValue();
String o1 = (String) sv1.get(CLINET_KEY + parkId);
log.info("删除客户端前数据：{}", o1);
Boolean f1 = ops1.delete(CLINET_KEY + parkId);
log.info("删除客户端标志：{}", f1);
String o2 = (String) sv1.get(CLINET_KEY + parkId);
log.info("删除客户端后数据：{}", o2);
redisSSTemplate.opsForValue().set(CLINET_KEY + parkId, parkId, 2, TimeUnit.MINUTES);*/
//redisSSTemplate.opsForValue().set(CLINET_KEY + parkId,parkId);
//存在就不删除也不设置,掉线之后两分钟内key过期后在设置，然后定时任务每一分钟内检查一次终端是否在线,恰好是key过期时间的1/2,所以key检测存,掉线时长大于1分钟可以被检测到。
} else {
redisSSTemplate.opsForValue().set(CLINET_KEY + parkId, parkId, 2, TimeUnit.MINUTES);
Boolean b = redisTemplate.opsForHash().hasKey(CLINET_ERROR_COUNT_KEY + parkId, "downLineCount");
if (b) {
redisTemplate.delete(CLINET_ERROR_COUNT_KEY + parkId);
log.info("删除掉线次数成功,parkId:{}", parkId);
}
}
} else {
log.info("parkid 为空");
//Tio.remove(channelContext, "parkid 为空!");
return httpResponse;
}
return httpResponse;
}
@Override
public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
log.info("==========握手成功=====");
}
@Override
public Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
return null;
}
@Override
public Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
log.info("==========关闭连接=====");
Tio.remove(channelContext, "receive close flag");
return null;
}
@Override
public Object onText(WsRequest wsRequest, String s, ChannelContext channelContext) throws Exception {
log.info("收到ws消息:{}", s);
if (StringUtil.isNotEmpty(s) && Objects.equals("007", JSON.parseObject(s).getString("code"))) {
log.info("心跳，心跳车场 : {} ", channelContext.userid);
//发送心跳消息给客户端
String msg = null;
//String msg = "{\"code\":\"008\",\"data\":\"你好,我是服务端！\"}";
return msg;
}
return null;
}
}
2.3 xxl-job定时任务扫描客户端是否在线然后发钉钉群告警消息
package com.dytz.barrier.gate.server.job;
import com.dytz.barrier.gate.server.config.ClientsConfig;
import com.dytz.barrier.gate.server.utils.DingDingUtil;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.util.Objects;
@Slf4j
@Component
public class ClientOnlineJobHandler {
@Autowired
private RedisTemplate redisSSTemplate;
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private ClientsConfig clientConfig;
private final String CLINET_KEY = "PARK:";
private final String CLINET_ERROR_COUNT_KEY = "PARK:ERROR:COUNT:";
@XxlJob("clientOnlineJobHandler")
public ReturnT<String> clientOnlineJobHandler(String param) {
if (clientConfig.getOpenFlag().equals("1")) {
XxlJobLogger.log("===================clientOnlineJobHandler====================");
for (String pkId : clientConfig.getClients()) {
if (redisSSTemplate.hasKey(CLINET_KEY + pkId)) {
String blrStr = String.valueOf(redisSSTemplate.opsForValue().get(CLINET_KEY + pkId));
if (StringUtils.isNotEmpty(blrStr)) {
log.info("clientId:{}在线", pkId);
XxlJobLogger.log("============在线设备pkId:{}===============", pkId);
continue;
}
}
redisTemplate.opsForHash().increment(CLINET_ERROR_COUNT_KEY + pkId, "downLineCount", 1);
if (this.checkFailCount(pkId)) {
XxlJobLogger.log("============疑似掉线设备pkId:{}===============", pkId);
StringBuffer sb = new StringBuffer("终端设备id【").append(pkId).append("】疑似掉线");
DingDingUtil.sendDingDingMarkdownGroupMsgAtAll("终端设备在线检测", sb.toString());
}
}
}
return ReturnT.SUCCESS;
}
/**
* 掉线次数统计限制发消息上限检查
*
* @param parkId
* @return
*/
private Boolean checkFailCount(String parkId) {
Boolean b = redisTemplate.opsForHash().hasKey(CLINET_ERROR_COUNT_KEY + parkId, "downLineCount");
if (b) {
Long count = (Long) redisTemplate.opsForHash().get(CLINET_ERROR_COUNT_KEY + parkId, "downLineCount");
log.info("=========failCount=========" + count);
if (Objects.nonNull(count) && count > Integer.valueOf(clientConfig.getErrorCount())) {
log.info("=========failCount大于10=========");
//不删除该key
//redisTemplate.delete(RefreshTokenConstants.BW_FAIL_COUNT + appNo);
return Boolean.FALSE;
}
}
return Boolean.TRUE;
}
}
3.客户端实现
3.1 客户端的pom依赖和yml配置
pom依赖
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.0modelVersion>
<parent>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-parentartifactId>
<version>2.7.2version>
<relativePath/>
parent>
<groupId>com.dytz.barrier.gate.clientgroupId>
<artifactId>barrier-gate-clientartifactId>
<version>0.0.1-SNAPSHOTversion>
<name>barrier-gate-clientname>
<packaging>jarpackaging>
<properties>
<java.version>18java.version>
<netty-all.version>4.1.79.Finalnetty-all.version>
properties>
<repositories>
<repository>
<id>nexusid>
<url>xxxxurl>
<releases>
<enabled>trueenabled>
<updatePolicy>alwaysupdatePolicy>
releases>
<snapshots>
<enabled>trueenabled>
<updatePolicy>alwaysupdatePolicy>
snapshots>
repository>
repositories>
<distributionManagement>
<repository>
<id>nexus-snapshotsid>
<name>Nexus snapshotsname>
<url>xxxxxxurl>
repository>
<snapshotRepository>
<id>nexus-snapshotsid>
<url>xxxxxxxxurl>
snapshotRepository>
distributionManagement>
<profiles>
<profile>
<id>uatid>
<properties>
<env>uatenv>
properties>
<activation>
<activeByDefault>trueactiveByDefault>
activation>
profile>
<profile>
<id>prodid>
<properties>
<env>prodenv>
properties>
profile>
profiles>
<dependencies>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-webartifactId>
dependency>
<dependency>
<groupId>org.projectlombokgroupId>
<artifactId>lombokartifactId>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-testartifactId>
<scope>testscope>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starterartifactId>
dependency>
<dependency>
<groupId>com.alibabagroupId>
<artifactId>fastjsonartifactId>
<version>1.2.83_noneautotypeversion>
dependency>
<dependency>
<groupId>cn.hutoolgroupId>
<artifactId>hutool-allartifactId>
<version>5.8.4version>
dependency>
<dependency>
<groupId>org.t-iogroupId>
<artifactId>tio-utilsartifactId>
<version>3.6.0.v20200315-RELEASEversion>
dependency>
<dependency>
<groupId>org.t-iogroupId>
<artifactId>tio-coreartifactId>
<version>3.6.0.v20200315-RELEASEversion>
dependency>
<dependency>
<groupId>org.t-iogroupId>
<artifactId>tio-websocket-clientartifactId>
<version>3.6.0.v20200315-RELEASEversion>
dependency>
dependencies>
<build>
<resources>
<resource>
<directory>src/main/resourcesdirectory>
<excludes>
<exclude>application-*.ymlexclude>
excludes>
resource>
<resource>
<directory>src/main/resourcesdirectory>
<filtering>truefiltering>
<includes>
<include>application.ymlinclude>
<include>application-${env}*.ymlinclude>
includes>
resource>
resources>
<plugins>
<plugin>
<groupId>org.apache.maven.pluginsgroupId>
<artifactId>maven-deploy-pluginartifactId>
<version>2.8.2version>
<configuration>
<skip>trueskip>
configuration>
plugin>
<plugin>
<groupId>org.apache.maven.pluginsgroupId>
<artifactId>maven-surefire-pluginartifactId>
<version>2.22.2version>
<configuration>
<skipTests>trueskipTests>
configuration>
plugin>
<plugin>
<groupId>org.apache.maven.pluginsgroupId>
<artifactId>maven-compiler-pluginartifactId>
<version>3.6.1version>
<configuration>
<source>1.8source>
<target>1.8target>
<encoding>UTF-8encoding>
configuration>
plugin>
<plugin>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-maven-pluginartifactId>
<version>2.2.9.RELEASEversion>
plugin>
plugins>
build>
project>
yml配置
fastjson:
parser:
safeMode: true
# 日记配置
logging:
level:
com.dytz.barrier.gate.client: debug
wsServerUrl: ws://127.0.0.1:19009/ws
3.2 客户端重试和心跳实现
package com.dytz.barrier.gate.client.ws;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dytz.barrier.gate.client.msg.SendMsg;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;
import org.tio.client.ClientChannelContext;
import org.tio.core.ChannelContext;
import org.tio.websocket.client.WebSocket;
import org.tio.websocket.client.WsClient;
import org.tio.websocket.client.config.WsClientConfig;
import org.tio.websocket.common.WsPacket;
import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
/**
* @author zlf
* @description:
* @time: 2022/8/19 15:24
*/
@Slf4j
@Configuration
public class MyWsClientConfig {
@Value("${wsServerUrl}")
private String wsServerUrl;
@Autowired
private ApplicationArguments applicationArguments;
private WsClient wsClient;
private ClientChannelContext cxt;
private WebSocket ws;
private static String parkId;
@PostConstruct
public void init() {
try {
Map<String, String> additionalHttpHeaders = new HashMap<>();
List<String> nonOptionArgs = applicationArguments.getNonOptionArgs();
if (!nonOptionArgs.isEmpty()) {
parkId = nonOptionArgs.get(0);
}
if (StringUtils.isEmpty(parkId)) {
throw new RuntimeException("请配置parkId之后在启动！");
}
log.info("parkId {}", parkId);
additionalHttpHeaders.put("parkid", parkId);
SendMsg sendMsg = new SendMsg();
sendMsg.setCode("2000");
sendMsg.setParkId(parkId);
String msg = JSON.toJSONString(sendMsg);
additionalHttpHeaders.put("parkid", "13");
WsClientConfig wsClientConfig = new WsClientConfig(
e -> {
log.info("emit open");
},
e -> {
WsPacket data = e.data;
if (Objects.nonNull(data)) {
String dataStr = data.getWsBodyText();
log.info("recv " + dataStr);
JSONObject jsonObject = JSON.parseObject(dataStr);
String code = jsonObject.getString("code");
String dataJson = jsonObject.getString("data");
log.info("收到服务端消息数据:{}", jsonObject.toJSONString());
if ("008".equals(code)) { //心跳数据包
log.info("收到心跳数据:{}", jsonObject.toJSONString());
} else { //非心跳数据
//业务处理
}
}
},
e -> {
log.info(String.format("emit close: %d, %s, %s", e.code, e.reason, e.wasClean));
wsClient.close();
wsClient = null;
},
e -> {
log.info(String.format("emit error: %s", e.msg));
wsClient.close();
wsClient = null;
},
Throwable::printStackTrace);
wsClient = WsClient.create(wsServerUrl, additionalHttpHeaders, wsClientConfig);
ws = wsClient.connect();
Thread thread = new Thread(() -> {
while (true) {
if (Objects.isNull(wsClient)) {
try {
wsClient = WsClient.create(wsServerUrl, additionalHttpHeaders, wsClientConfig);
ws = wsClient.connect();
} catch (Exception e) {
e.printStackTrace();
try {
Thread.sleep(1000L);
} catch (InterruptedException e2) {
e2.printStackTrace();
}
try {
ws = wsClient.connect();
} catch (Exception e1) {
e1.printStackTrace();
}
}
}
if (Objects.isNull(wsClient)) {
continue;
}
cxt = wsClient.getClientChannelContext();
if (Objects.isNull(cxt)) {
continue;
}
ChannelContext.CloseCode closeCode = cxt.getCloseCode();
if (2 == closeCode.getValue() || cxt.isClosed || cxt.isRemoved) {
wsClient.close();
wsClient = null;
if (Objects.isNull(wsClient)) {
continue;
}
}
if (Objects.nonNull(ws)) {
//发送心跳
ws.send("{\"code\":\"007\",\"data\":\"你好,我是客户端！\"}");
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
thread.start();
} catch (Exception e) {
e.printStackTrace();
wsClient.close();
wsClient = null;
}
}
}
4.客户端和服务端的demo分享
链接：https://pan.baidu.com/s/16jlic4LEu5RE7-nMsWfupA
提取码：0965