[v0.0.0] 原始哈勃服务器代码

master
shiyi 2 months ago
commit f016e02e75

2
.gitignore vendored

@ -0,0 +1,2 @@
/target/
/.idea/

@ -0,0 +1,132 @@
<?xml 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.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.5.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.platform</groupId>
<artifactId>pass-through</artifactId>
<version>1.0.0</version>
<name>pass-through</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.32.Final</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.76</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- log4j2 日志记录-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>
<!-- 加上这个才能辨认到log4j2.yml文件 -->
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-yaml</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.7.0</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.7.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<!--<plugin>
<groupId>org.mybatis.generator</groupId>
<artifactId>mybatis-generator-maven-plugin</artifactId>
<version>1.3.2</version>
<executions>
<execution>
<id>Generate MyBatis Artifacts</id>
<goals>
<goal>generate</goal>
</goals>
</execution>
</executions>
<configuration>
&lt;!&ndash; generator 工具配置文件的位置 &ndash;&gt;
<configurationFile>src/main/resources/mybatis-generator/generatorConfig.xml</configurationFile>
<verbose>true</verbose>
<overwrite>true</overwrite>
</configuration>
<dependencies>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.34</version>
</dependency>
<dependency>
<groupId>org.mybatis.generator</groupId>
<artifactId>mybatis-generator-core</artifactId>
<version>1.3.2</version>
</dependency>
</dependencies>
</plugin>-->
</plugins>
</build>
</project>

@ -0,0 +1,14 @@
package com.platform;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class BootApplication {
public static void main(String[] args) {
SpringApplication.run(BootApplication.class, args);
}
}

@ -0,0 +1,29 @@
package com.platform.config;
import org.springframework.stereotype.Component;
import javax.servlet.*;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
@Component
public class ClientHttpFilter implements Filter {
@Override
public void doFilter(ServletRequest req, ServletResponse res, FilterChain chain) throws IOException, ServletException {
HttpServletResponse response = (HttpServletResponse) res;
response.setHeader("Access-Control-Allow-Origin", "*");
response.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS, DELETE");
response.setHeader("Access-Control-Max-Age", "3600");
response.setHeader("Access-Control-Allow-Headers", "x-requested-with");
chain.doFilter(req, res);
}
@Override
public void init(FilterConfig filterConfig) {
}
@Override
public void destroy() {
}
}

@ -0,0 +1,14 @@
package com.platform.config;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
@Configuration
@Data
public class ServiceConfig {
@Value("${log.debug}")
private boolean debug;
@Value("${service.port}")
private Integer port;
}

@ -0,0 +1,33 @@
package com.platform.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;
@Configuration
@EnableAsync
public class ThreadPoolConfig {
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 设置核心线程数
executor.setCorePoolSize(8);
// 设置最大线程数
executor.setMaxPoolSize(100);
// 设置队列容量
executor.setQueueCapacity(8);
// 设置线程活跃时间(秒)
executor.setKeepAliveSeconds(300);
// 设置默认线程名称
executor.setThreadNamePrefix("thread-");
// 设置拒绝策略rejection-policy当pool已经达到max size的时候如何处理新任务 CALLER_RUNS不在新线程中执行任务而是有调用者所在的线程来执行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 等待所有任务结束后再关闭线程池
executor.setWaitForTasksToCompleteOnShutdown(true);
return executor;
}
}

@ -0,0 +1,219 @@
package com.platform.controller;
import com.platform.model.ManyToMany;
import com.platform.model.OneToMany;
import com.platform.util.SessionCache;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiModelProperty;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import java.util.*;
import java.util.stream.Collectors;
@RestController
@Slf4j
public class ClientController {
@Autowired
private SessionCache sessionCache;
/**
* session ip
* @return
*/
@GetMapping("/session/ips")
public List<String> getSessions() {
List<String> ips = new ArrayList<>();
Set set= sessionCache.tcpSession.keySet();
Iterator iterator = set.iterator();
while (iterator.hasNext()){
ips.add(iterator.next().toString());
}
return ips;
}
/**
*
*
* @param source
* @param target
* @return
*/
@PostMapping("/mapping/oneToOne")
public Boolean addSnMap(@RequestParam("source") String source, @RequestParam("target") String target) {
sessionCache.snToSnMap.put(source, target);
sessionCache.snToSnMap.put(target, source);
return true;
}
/**
*
* @param param
* @return
*/
@PostMapping("/mapping/oneToMany")
public Boolean addSnsMap(@RequestBody OneToMany param) {
List<String> list = param.getTarget();
List<String> target = list.stream().map(String::toUpperCase).collect(Collectors.toList());
String source = param.getSource().toUpperCase();
target.forEach(sn -> sessionCache.snToSnMap.put(sn, source));
if(!sessionCache.snList.contains(source)) {
sessionCache.snList.add(source);
}
List<String> oldTarget = sessionCache.snToListMap.get(source);
if (!ObjectUtils.isEmpty(oldTarget)) {
oldTarget.forEach(sn ->{
if (!target.contains(sn)) {
target.add(sn);
}
});
}
sessionCache.snToListMap.put(source, target);
return true;
}
/**
*
* @param param
* @return
*/
@PostMapping("/mapping/all")
public Boolean addSourcesAndTargetsMap(@RequestBody ManyToMany param) {
List<String> targetList = param.getTarget().stream().map(String::toUpperCase).collect(Collectors.toList());
List<String> sourceList = param.getSource().stream().map(String::toUpperCase).collect(Collectors.toList());
List<String> sourceTmp = new ArrayList<>();
targetList.forEach(target->{
List<String> oldSource = sessionCache.mappingListMap.get(target);
if (!ObjectUtils.isEmpty(oldSource)) {
oldSource.forEach(sn ->{
if (!sourceList.contains(sn)) {
sourceList.add(sn);
}
});
}
List<String> newSource= new ArrayList<>();
newSource.addAll(sourceList);
if (!sourceTmp.isEmpty()){
newSource.addAll(sourceTmp);
sourceTmp.clear();
}
sessionCache.mappingListMap.put(target, newSource);
});
List<String> targetTmp = new ArrayList<>();
sourceList.forEach(source ->{
List<String> oldTarget = sessionCache.mappingListMap.get(source);
if (!ObjectUtils.isEmpty(oldTarget)) {
oldTarget.forEach(sn ->{
if (!targetList.contains(sn)) {
targetTmp.add(sn);
}
});
}
List<String> newTarget = new ArrayList<>();
newTarget.addAll(targetList);
if (!targetTmp.isEmpty()){
newTarget.addAll(targetTmp);
targetTmp.clear();
}
sessionCache.mappingListMap.put(source, newTarget);
});
return true;
}
/**
*
* @param param
* @return
*/
@PostMapping("/mapping/delete")
public Boolean deleteSourcesAndTargetsMap(@RequestBody ManyToMany param) {
List<String> sourceList = param.getSource().stream().map(String::toUpperCase).collect(Collectors.toList());
List<String> targetList = param.getTarget().stream().map(String::toUpperCase).collect(Collectors.toList());
List<String> all = new ArrayList<>();
all.addAll(sourceList);
all.addAll(targetList);
all.forEach(code ->{
List<String> valueList = sessionCache.mappingListMap.get(code);
List<String> deletedList = new ArrayList<>();
if (!ObjectUtils.isEmpty(valueList)) {
valueList.forEach(value ->{
if (all.contains(value)){
deletedList.add(value);
}
});
deletedList.forEach(deleted -> valueList.remove(deleted));
if (valueList.size() > 0) {
sessionCache.mappingListMap.put(code, valueList);
} else{
sessionCache.mappingListMap.remove(code);
}
}
});
return true;
}
/**
*
* @return
*/
@GetMapping("/mapping/oneToOne")
public Map<String, String> getSnMap() {
return sessionCache.snToSnMap;
}
/**
*
* @return
*/
@GetMapping("/mapping/oneToMany")
public Map<String, List<String>> getSnListMap() {
return sessionCache.snToListMap;
}
/**
*
* @return
*/
@GetMapping("/mapping/get")
public Map<String, List<String>> getTargetListMap() {
return sessionCache.mappingListMap;
}
/**
*
*
* @param supervisor
* @param monitored
* @return
*/
@ApiImplicitParams({
@ApiImplicitParam(name = "supervisor", value = "监控者", required = true),
@ApiImplicitParam(name = "monitored", value = "被监控者", required = true),
})
@GetMapping("/sn/mapping/monitor/add")
@ApiModelProperty(value = "配置监控接口")
public Boolean addMonitorMap(String supervisor, String monitored) {
if(StringUtils.isEmpty(supervisor) || StringUtils.isEmpty(monitored)){
return false;
}
sessionCache.snToMonitorMap.clear();
sessionCache.snToMonitorMap.put(monitored,supervisor);
return true;
}
/**
*
* @return
*/
@GetMapping("/sn/mapping/monitor/get")
@ApiModelProperty(value = "查询监控关系接口")
public Map<String,String> getMonitorMap() {
return sessionCache.snToMonitorMap;
}
}

@ -0,0 +1,12 @@
package com.platform.model;
import lombok.Data;
import java.io.Serializable;
import java.util.List;
@Data
public class ManyToMany implements Serializable {
private List<String> source;
private List<String> target;
}

@ -0,0 +1,12 @@
package com.platform.model;
import lombok.Data;
import java.io.Serializable;
import java.util.List;
@Data
public class OneToMany implements Serializable {
private String source;
private List<String> target;
}

@ -0,0 +1,288 @@
package com.platform.service;
import com.platform.config.ServiceConfig;
import com.platform.util.CRCUtil;
import com.platform.util.ControlDevice;
import com.platform.util.SessionCache;
import com.platform.util.StringUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
@Slf4j
public class InMessageHandler extends ChannelInboundHandlerAdapter {
private boolean debug;
private StringUtil stringUtil;
private SessionCache sessionCache;
private static final int head1 = 0xAA;
private static final int head2 = 0x44;
private static final int head3 = 0x61;
private static final int lengthLow = 0x03;
private static final int lengthHigh = 0x00;
public InMessageHandler() {
}
public InMessageHandler(SessionCache sessionCache, StringUtil stringUtil, ServiceConfig config) {
this.stringUtil = stringUtil;
this.sessionCache = sessionCache;
this.debug = config.isDebug();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
log.error("【系统异常】======>", cause);
ctx.close();
}
/**
*
* 5;
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
if (obj instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) obj;
IdleState state = event.state();
if (state == IdleState.ALL_IDLE) {
//设备离线,更改设备状态
log.warn("【客户端" + ctx.channel().remoteAddress() + "进入idle状态】");
ChannelFuture future = ctx.channel().close();
if (!future.isSuccess()) {
log.warn("【客户端异常关闭】", future.cause());
}
}
}
super.userEventTriggered(ctx, obj);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg0) throws Exception {
String address = getAddress(ctx);
byte[] msg = (byte[])msg0;
Byte fkID = msg[4];
//上线握手阶段
if (!sessionCache.addToSnMap.containsKey(address)) { // 判断连接是否已经建立,如未建立则先握手
// 握手数据不满足协议长度,立即返回
String d = stringUtil.bytesToHexString(msg);
log.info("【接收握手原数据】 " + d);
// byte[] crc = CRC.crc16(msg, 2, 16);
// if (crc[0] != msg[18] || crc[1] != msg[19]){
// log.warn("握手数据crc校验错误期望crc " + stringTool.bytesToHexString(crc) + ", 实际收到 " + d);
// protocolAck(ctx,(byte) 0);
// return;
// }
// 数据类型 1为地面站 2为终端设备
String type = new String(msg, 5, 1, Charset.defaultCharset());
// deviceCode 终端设备取后六位
String deviceCode;
if(type.equals("1")) {
// 首先读取deviceCode 哈勃deviceCode是前面6个byte
deviceCode = new String(msg, 6, 12, Charset.defaultCharset()).toUpperCase(); // 地面站
} else if (type.equals("2")){
deviceCode = new String(msg, 12, 6 , Charset.defaultCharset()).toUpperCase(); // 哈勃
} else {
protocolAck(ctx,(byte) 0);
return;
}
log.info("【读取握手数据】======> deviceCode = " + deviceCode +" type = " + type);
//删除旧的连接。如果存在异常断开的时候,旧连接并未清除。这里在上线的时候检查并清除
String addr0 = sessionCache.snToAddMap.get(deviceCode);
if (Objects.nonNull(addr0)) {
sessionCache.tcpSession.remove(addr0);
sessionCache.addToSnMap.remove(addr0);
log.info("【握手删除旧连接】====> " + addr0 + "(" + deviceCode + ")删除");
}
//添加新连接
sessionCache.addToSnMap.put(address, deviceCode); // ip-mac
sessionCache.snToAddMap.put(deviceCode, address); // mac- ip
sessionCache.tcpSession.put(address, ctx.channel()); // ip - channel
protocolAck(ctx,(byte) 1);
return; //握手结束立即返回
}else{ // 如果已经建立连接,判断是否是地面站的控制信息
if(msg[0]==(byte)0xCC&msg[1]==(byte)0x06&msg[2]==(byte)0x04){
byte[] content = new byte[10];
content[0] = (byte)0x01;
int uavNum = StringUtil.byteToInt(msg[3]);
String source = sessionCache.addToSnMap.get(address); //mac
ControlDevice.clearCurrenCtrDeviceByMac(source);
int errNum = 0;
// List<String> deviceSns = new ArrayList<>();
for(int i=0;i<uavNum;i++){
try{
String deviceSn= ControlDevice.getSn(msg[4+i]); // 根据fkId获取哈勃序列号
log.warn("当前"+address+"在控无人机更新为:"+ deviceSn);
ControlDevice.currenCtrDevice.put(deviceSn,source); //mac - 当前mac控制的设备编号
}catch (Exception e){
errNum=errNum+1;
content[0] = (byte)0x00;
content[1] = (byte)errNum;
content[errNum+1] = msg[4+i];
}
// deviceSns.add(deviceSn);
}
response(ctx,content);
return;
}
}
// 已经建立连接且非地面站控制信息,则进行数据转发
if (debug) {
log.info("【转发数据】======> " + stringUtil.bytesToHexString(msg));
}
String sn = sessionCache.addToSnMap.get(address); //sn =mac
//监控相关
if(!sessionCache.snToMonitorMap.isEmpty()){
Optional<Channel> sendChannelOpt = sessionCache.findMonitorChannel (address);
if (sendChannelOpt.isPresent()) {
Channel sendChannel = sendChannelOpt.get();
log.info("(监控)out channel: " + sendChannel.remoteAddress()+"(" + sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/",""))
+ ") 即将发送 --- " + "in channel: " + ctx.channel().remoteAddress()+"(" + sn + ")发来的TCP消息");
sendChannel.writeAndFlush(msg0);
}
}
if (!sessionCache.mappingListMap.isEmpty()){
if (sessionCache.mappingListMap.get(sn)!=null){
List<Channel> sendChannelOpt = sessionCache.findMappingTargetChannel(sn); // 设备编号对应的通道
log.info("send channel count = " + sendChannelOpt.size());
sendChannelOpt.forEach(sendChannel -> {
log.info("out channel: " + sendChannel.remoteAddress()+"(" + sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/",""))
+ ")即将发送 --- " + "in channel: " + ctx.channel().remoteAddress()+"(" + sn + ")发来的TCP消息");
String sendIp = sessionCache.findSnByChannel(sendChannel); // 通道找到IP
String sendSn = sessionCache.addToSnMap.get(sendIp); // IP找到设备编号
if(sendSn.length()==6){
List<String> devices = ControlDevice.getCurrenCtrDevicesByMac(sn);
if(devices.contains(sendSn)||devices.contains("all")){
sendChannel.writeAndFlush(msg0);
}
}
if(sendSn.length()==12) {
sendChannel.writeAndFlush(msg0);
}
});
return;
}
}
/* String sn = global.addToSnMap.get(address);
if (!global.snList.isEmpty()){
if (global.snList.contains(sn)){
List<Channel> sendChannelOpt = global.findManyTargetChannel(sn);
log.info("send channel count = " + sendChannelOpt.size());
sendChannelOpt.forEach(sendChannel -> {
log.info("out channel: " + sendChannel.remoteAddress()+"(" + global.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/",""))
+ ")即将发送 --- " + "in channel: " + ctx.channel().remoteAddress()+"(" + sn + ")发来的TCP消息");
sendChannel.writeAndFlush(msg0);
});
return;
}
}*/
/*//转发数据
Optional<Channel> sendChannelOpt = global.findTargetChannel(address);
if (!sendChannelOpt.isPresent()) {
log.warn("无有效连接, address = " + address);
} else {
Channel sendChannel = sendChannelOpt.get();
log.info("out channel: " + sendChannel.remoteAddress()+"(" + global.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/",""))
+ ") 即将发送 --- " + "in channel: " + ctx.channel().remoteAddress()+"(" + sn + ")发来的TCP消息");
sendChannel.writeAndFlush(msg0);
}*/
super.channelRead(ctx, msg0);
}
private void protocolAck(ChannelHandlerContext ctx, byte data){
byte[] ack = new byte[8];
ack[0] = (byte) head1;
ack[1] = (byte) head2;
ack[2] = (byte) head3;
ack[3] = (byte) lengthLow;
ack[4] = (byte) lengthHigh;
ack[5] = data;
byte[] crc = CRCUtil.crc16(ack, 2, 4);
ack[6] = crc[0];
ack[7] = crc[1];
ctx.channel().writeAndFlush(ack);
}
private void response(ChannelHandlerContext ctx,byte[] content){
ByteBuf buf = Unpooled.buffer(content.length+4);
buf.writeByte((byte) 0xcc);
buf.writeByte((byte) 0x06);
buf.writeByte((byte) 0x04);
buf.writeByte((byte) 0x01);
buf.writeBytes(content);
//ack[4] = (byte) 0x01;
ctx.channel().writeAndFlush(buf);
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
log.warn("【设备上线】====> " + getAddress(ctx) + "上线");
sessionCache.tcpSession.put(getAddress(ctx), ctx.channel());
super.handlerAdded(ctx);
}
private String getAddress(ChannelHandlerContext ctx) {
String address = ctx.channel().remoteAddress().toString().toUpperCase();
return address.replace("/", "");
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
String addr = getAddress(ctx);
String sn = sessionCache.addToSnMap.get(addr);
log.warn("【设备离线】====> " + addr + "(" + sn + ")离线");
ControlDevice.clearCurrenCtrDeviceByMac(sn);
sessionCache.tcpSession.remove(addr);
sessionCache.addToSnMap.remove(addr);
if (!StringUtils.isEmpty(sn)) {
// 只有当要被删除的地址跟snToAddMap里面的地址匹配的时候才删除
// 因为存在一种情况同一个sn上线了snToAddMap数据已经被正确的地址覆盖此时不应该删除
if (addr.equals(sessionCache.snToAddMap.get(sn))) {
sessionCache.snToAddMap.remove(sn);
}
}
super.handlerRemoved(ctx);
}
}

@ -0,0 +1,75 @@
package com.platform.service;
import com.platform.config.ServiceConfig;
import com.platform.util.SessionCache;
import com.platform.util.StringUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class ServerService {
public ChannelFuture mChannelFuture;
@Autowired
private ServiceConfig config;
@Autowired
private StringUtil stringUtil;
@Autowired
private SessionCache sessionCache;
private void startServer() {
//服务端需要2个线程组 boss处理客户端连接 work进行客服端连接之后的处理
log.warn("【读取配置端口】 端口 = " + config.getPort());
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup work = new NioEventLoopGroup(10);
try {
ServerBootstrap bootstrap = new ServerBootstrap();
//服务器 配置
bootstrap.group(boss, work);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel channel) throws Exception {
channel.pipeline().addLast(new ByteArrayDecoder());
channel.pipeline().addLast(new ByteArrayEncoder());
channel.pipeline().addLast(new IdleStateHandler(0, 0,
60 * 24, TimeUnit.MINUTES));
channel.pipeline().addLast(new InMessageHandler(sessionCache, stringUtil, config));
}
});
bootstrap.option(ChannelOption.SO_BACKLOG, 2048);
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
mChannelFuture = bootstrap.bind(config.getPort()).sync();
mChannelFuture.channel().closeFuture().sync();
log.warn("【服务器启动成功========端口:" + config.getPort() + "】");
} catch (Exception e) {
log.error("【服务器启动失败】", e);
} finally {
//关闭资源
boss.shutdownGracefully();
work.shutdownGracefully();
}
}
@PostConstruct()
public void init() {
log.warn("【初始化......】");
//需要开启一个新的线程来执行netty server 服务器
new Thread(() -> startServer()).start();
}
}

@ -0,0 +1,102 @@
package com.platform.util;
public class CRCUtil {
/**
* 8
*/
private static final int BITS_OF_BYTE = 8;
/**
*
*/
private static final int POLYNOMIAL = 0xa001;
/**
*
*/
private static final int INITIAL_VALUE = 0xffff;
/**
* CRC16
*
* @param bytes
* @return
*/
public static byte[] crc16(byte[] bytes, int offset, int length) {
int res = INITIAL_VALUE;
for (int j = offset; j < offset + length; j++) {
res = res ^ bytes[j];
for (int i = 0; i < BITS_OF_BYTE; i++) {
res = (res & 0x0001) == 1 ? (res >> 1) ^ POLYNOMIAL : res >> 1;
}
}
return toLHHex(res);
}
/**
* 16
*
* @param src
* @return
*/
private static int revert(int src) {
int lowByte = (src & 0xFF00) >> 8;
int highByte = (src & 0x00FF) << 8;
return lowByte | highByte;
}
/**
* 16
*
* @param src
* @return ,
*/
public static byte[] toLHHex(int src) {
byte[] b = new byte[2];
b[0] = (byte) (src & 0xff);
b[1] = (byte) (src >> 8 & 0xff);
return b;
}
private static final int b0 = 0xAA;
private static final int b1 = 0x44;
private static final int b2 = 0x51;
private static final int b3 = 0x15;
private static final int b4 = 0x00;
private static final int b5 = 0x31;
private static final int b6 = 0x30;
private static final int b7 = 0x30;
private static final int b8 = 0x30;
private static final int b9 = 0x31;
private static final int b10 = 0x37;
private static final int b11 = 0x62;
private static final int b12 = 0x64;
private static final int b13 = 0x34;
private static final int b14 = 0x32;
private static final int b15 = 0x65;
private static final int b16 = 0x36;
private static final int b17 = 0x32;
public static void main(String[] args) {
byte[] ack = new byte[18];
ack[0] = (byte)b0;
ack[1] = (byte)b1;
ack[2] = (byte)b2;
ack[3] = (byte)b3;
ack[4] = (byte)b4;
ack[5] = (byte)b5;
ack[6] = (byte)b6;
ack[7] = (byte)b7;
ack[8] = (byte)b8;
ack[9] = (byte)b9;
ack[10] = (byte)b10;
ack[11] = (byte)b11;
ack[12] = (byte)b12;
ack[13] = (byte)b13;
ack[14] = (byte)b14;
ack[15] = (byte)b15;
ack[16] = (byte)b16;
ack[17] = (byte)b17;
byte[] b = crc16(ack, 2, 16);
StringUtil tool = new StringUtil();
System.out.println(tool.bytesToHexString(b));
}
}

@ -0,0 +1,69 @@
package com.platform.util;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class ControlDevice {
private static final HashMap<Byte, String> deviceSnToFkID = new HashMap<Byte, String>(){{
put((byte) 0x11,"D80A72");
put((byte) 0x13,"D80B00");
put((byte) 0x23,"D80AD8");
put((byte) 0x24,"D80A83");
put((byte) 0x25,"D80A87");
put((byte) 0x30,"D80A88");
put((byte) 0x31,"D80B0A");
put((byte) 0x32,"D80AB6");
put((byte) 0x01,"F492C1");
put((byte) 0x02,"F492C2");
put((byte) 0x03,"F492C3");
put((byte) 0xff,"all");
}};
public static String getSn(Byte fkID){return deviceSnToFkID.get(fkID);}
public static boolean isTure(Byte fkID){return deviceSnToFkID.containsKey(fkID);}
public static int getFk(String uavId){
int fk = 0;
for(Byte key: deviceSnToFkID.keySet()){
if(deviceSnToFkID.get(key).equals(uavId)){
fk = key;
}
}
return fk;
}
public static Map<String, String> currenCtrDevice = new ConcurrentHashMap<>(16); // 设备编号对应的mac地址
public static void clearCurrenCtrDeviceByMac(String mac){
for(String key: currenCtrDevice.keySet()){
if(currenCtrDevice.get(key).equals(mac)){
currenCtrDevice.remove(key);
}
}
}
public static List<String> getCurrenCtrDevicesByMac(String mac){
List<String> ctrDevices = new ArrayList<>();
for(String key: currenCtrDevice.keySet()){
if(currenCtrDevice.get(key).equals(mac)){
ctrDevices.add(key);
}
}
return ctrDevices;
}
}

@ -0,0 +1,143 @@
package com.platform.util;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@Component
@Slf4j
public class SessionCache {
public Map<String, Channel> tcpSession = new ConcurrentHashMap<>(16);
public Map<String, String> addToSnMap = new ConcurrentHashMap<>(16); // ip地址对应的设备编号
public Map<String, String> snToAddMap = new ConcurrentHashMap<>(16); // 设备编号对应的ip地址
public Map<String, String> snToSnMap = new ConcurrentHashMap<>(16); //一对一映射关系
public Map<String, String> snToMonitorMap = new ConcurrentHashMap<>(16); //监控映射
public Map<String, List<String>> mappingListMap = new ConcurrentHashMap<>(16); //所有映射关系
public Map<String, List<String>> snToListMap = new ConcurrentHashMap<>(16); //一对多映射关系
public List<String> snList = new ArrayList<>();
public Optional<Channel> findTargetChannel(String sourceAddr) {
String sSn = addToSnMap.get(sourceAddr);
String tSn = snToSnMap.get(sSn);
if (tSn == null) {
log.warn("找不到" + sourceAddr + "的目标映射;" +
"addToSnMap = " + JSON.toJSONString(addToSnMap) + "; " +
"snToSnMap = " + JSON.toJSONString(snToSnMap));
return Optional.empty();
}
String targetAddr = snToAddMap.get(tSn);
if (targetAddr == null) {
log.warn("目标映射" + tSn + "的地址找不到; " +
"snToAddMap = " + JSON.toJSONString(snToAddMap));
return Optional.empty();
}
Channel c = tcpSession.get(targetAddr);
if (c == null) {
log.warn("目标地址" + targetAddr + "的session channel找不到; " +
"tcpSession = " + JSON.toJSONString(tcpSession));
return Optional.empty();
}
return Optional.of(c);
}
public Optional<Channel> findMonitorChannel(String sourceAddr) {
String sSn = addToSnMap.get(sourceAddr);
String tSn = snToMonitorMap.get(sSn);
if (tSn == null) {
log.warn("找不到" + sourceAddr + "的目标映射;" +
"addToSnMap = " + JSON.toJSONString(addToSnMap) + "; " +
"snToMonitorMap = " + JSON.toJSONString(snToMonitorMap));
return Optional.empty();
}
String targetAddr = snToAddMap.get(tSn);
if (targetAddr == null) {
log.warn("目标映射" + tSn + "的地址找不到; " +
"snToAddMap = " + JSON.toJSONString(snToAddMap));
return Optional.empty();
}
Channel c = tcpSession.get(targetAddr);
if (c == null) {
log.warn("目标地址" + targetAddr + "的session channel找不到; " +
"tcpSession = " + JSON.toJSONString(tcpSession));
return Optional.empty();
}
return Optional.of(c);
}
public List<Channel> findManyTargetChannel(String sSn) {
List<Channel> channel = new ArrayList<>();
List<String> tSn = snToListMap.get(sSn);
if (!ObjectUtils.isEmpty(tSn)) {
tSn.forEach(s -> {
String targetAddr = snToAddMap.get(s);
if (targetAddr != null) {
Channel c = tcpSession.get(targetAddr);
if (c == null) {
log.warn("targetAddr " + targetAddr + " channel is null" + ", tcpSession = " + JSON.toJSONString(tcpSession));
} else {
channel.add(c);
}
}
});
}
return channel;
}
public List<Channel> findMappingTargetChannel(String sSn) {
List<Channel> channel = new ArrayList<>();
List<String> tSn = mappingListMap.get(sSn); // 设备的映射关系
if (!ObjectUtils.isEmpty(tSn)) {
tSn.forEach(s -> { //
String targetAddr = snToAddMap.get(s); // 根据设备编号找到ip地址
if (targetAddr != null) {
Channel c = tcpSession.get(targetAddr); // 根据ip找到通道
if (c == null) {
log.warn("targetAddr " + targetAddr + " channel is null" + ", tcpSession = " + JSON.toJSONString(tcpSession));
} else {
channel.add(c);
}
}
});
}
return channel;
}
/**
* channelip
*/
public String findSnByChannel(Channel channel){
String fk = null;
for(String key: tcpSession.keySet()){
if(tcpSession.get(key).equals(channel)){
fk = key;
}
}
return fk;
}
}

@ -0,0 +1,390 @@
package com.platform.util;
import org.springframework.stereotype.Component;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.regex.Pattern;
@Component
public class StringUtil {
private static Pattern pattern = Pattern.compile("\\b\\w+:\\w+:\\w+:\\w+:\\w+:\\w+\\b");
/**
* Convert the hex string to byte array;
*
* @param strHexValue the hex string value.
* @return byte[] the resolved value.
*/
public byte[] stringToByteArray(String strHexValue) {
String[] strAryHex = strHexValue.split(" ");
byte[] btAryHex = new byte[strAryHex.length];
try {
int nIndex = 0;
for (String strTemp : strAryHex) {
btAryHex[nIndex] = (byte) Integer.parseInt(strTemp, 16);
nIndex++;
}
} catch (NumberFormatException e) {
}
return btAryHex;
}
/**
* Convert the Hex string array to byte array
*
* @param strAryHex the hex string value.
* @param nLen the needed parse length.
* @return byte[] the resolved value.
*/
public byte[] stringArrayToByteArray(String[] strAryHex, int nLen) {
if (strAryHex == null) {
return null;
}
if (strAryHex.length < nLen) {
nLen = strAryHex.length;
}
byte[] btAryHex = new byte[nLen];
try {
for (int i = 0; i < nLen; i++) {
btAryHex[i] = (byte) Integer.parseInt(strAryHex[i], 16);
}
} catch (NumberFormatException e) {
}
return btAryHex;
}
/**
* Convert the Hex string array to byte array
*
* @param btAryHex the hex string value.
* @param nIndex the starting parse position.
* @param nLen the needed parse length.
* @return byte[] the resolved value.
*/
public String byteArrayToString(byte[] btAryHex, int nIndex, int nLen) {
if (nIndex + nLen > btAryHex.length) {
nLen = btAryHex.length - nIndex;
}
String strResult = String.format("%02X", btAryHex[nIndex]);
for (int nloop = nIndex + 1; nloop < nIndex + nLen; nloop++) {
String strTemp = String.format(" %02X", btAryHex[nloop]);
strResult += strTemp;
}
return strResult;
}
public String bytesToHexString(byte[] bArray) {
StringBuffer sb = new StringBuffer(bArray.length);
String sTemp;
for (int i = 0; i < bArray.length; i++) {
sTemp = Integer.toHexString(0xFF & bArray[i]);
if (sTemp.length() < 2) {
sb.append(0);
}
sb.append(sTemp.toUpperCase());
}
return sb.toString();
}
/**
* Convert the String to StringArray
*
* @param strValue the src string
* @param nLen needed convert length
* @return String[] the parsed value.
*/
public String[] stringToStringArray(String strValue, int nLen) {
String[] strAryResult = null;
if (strValue != null && !"".equals(strValue)) {
ArrayList<String> strListResult = new ArrayList<String>();
String strTemp = "";
int nTemp = 0;
for (int nloop = 0; nloop < strValue.length(); nloop++) {
if (strValue.charAt(nloop) == ' ') {
continue;
} else {
nTemp++;
if (!pattern.matcher(strValue.substring(nloop, nloop + 1)).matches()) {
return strAryResult;
}
strTemp += strValue.substring(nloop, nloop + 1);
if ((nTemp == nLen) || (nloop == strValue.length() - 1
&& (strTemp != null && !"".equals(strTemp)))) {
strListResult.add(strTemp);
nTemp = 0;
strTemp = "";
}
}
}
if (strListResult.size() > 0) {
strAryResult = new String[strListResult.size()];
for (int i = 0; i < strAryResult.length; i++) {
strAryResult[i] = strListResult.get(i);
}
}
}
return strAryResult;
}
/**
* Convert the byte array to special encode char array.
*
* @param bytes the needed byte array.
* @param encoding the encoding format.
* @return char[] the converted char array.
*/
@Deprecated
public char[] getChars(byte[] bytes, String encoding) {
Charset cs = Charset.forName(encoding);
ByteBuffer bb = ByteBuffer.allocate(bytes.length);
bb.put(bytes);
bb.flip();
CharBuffer cb = cs.decode(bb);
return cb.array();
}
/**
* resolve the bytes String data to ASCII.String like this "2f3Eab"
*
* @param hexString the needed parse data.
* @return String the return value.
*/
@Deprecated
public String toASCIIString(String hexString) {
StringBuilder stringBuilder = new StringBuilder();
int i = 0;
while (i < hexString.length()) {
stringBuilder.append((char) Integer.parseInt(hexString.substring(i, i + 2), 16));
i = i + 2;
}
return stringBuilder.toString();
}
/**
* Resolve the bytes String data to ASCII.String like this "2f3Eab"
*
* @param hexString the needed parse data.
* @return String the return value.
*/
public String hexStringToASCIIString(String hexString) {
StringBuilder stringBuilder = new StringBuilder();
if (hexString == null || "".equals(hexString)) {
return null;
}
for (int i = 0; i < hexString.length(); i = i + 2) {
char high = hexString.charAt(i);
char low = hexString.charAt(i + 1);
char no = (char) (charToInt(high) * 16 + charToInt(low));
stringBuilder.append(no);
}
return stringBuilder.toString();
}
/**
* Convert the char 1-f to int value;
*
* @param c the needed converter character.
* @return int the converted value.
*/
private int charToInt(char c) {
int num = -1;
num = "0123456789ABCDEF".indexOf(String.valueOf(c));
if (num < 0) {
num = "0123456789abcdef".indexOf(String.valueOf(c));
}
return num;
}
/**
* Get the sub bytes of the parent bytes;
*
* @param bytes parent bytes;
* @param start start position;
* @param end end position;
* @return the child bytes;
*/
public byte[] subBytes(byte[] bytes, int start, int end) {
byte[] subBytes = new byte[end - start];
System.arraycopy(bytes, start, subBytes, 0, end - start);
return subBytes;
}
/**
* Get the child position in their parents string
*
* @param parentBytes parent bytes;
* @param childBytes child bytes;
* @param startPosition start scan position;
* @return if -1 child not in parent string,or the child first position found in parent;
*/
public int subBytesContains(byte[] parentBytes, byte[] childBytes, int startPosition) {
if (parentBytes.length < childBytes.length) {
return -1;
}
for (int i = startPosition; i < parentBytes.length; i++) {
for (int j = 0; j < childBytes.length; j++) {
if (parentBytes[i + j] == childBytes[j]) {
if (j == childBytes.length - 1) {
return i;
}
continue;
} else {
break;
}
}
}
return -1;
}
/**
* Convert hexString to byte array.
*
* @param hexString to bytes
* @return The converted byte array.
*/
public byte[] hexStringToBytes(String hexString) {
byte[] bytes = new byte[hexString.length() / 2];
if (hexString == null || "".equals(hexString)) {
return null;
}
for (int i = 0; i < hexString.length(); i = i + 2) {
char high = hexString.charAt(i);
char low = hexString.charAt(i + 1);
bytes[i / 2] = (byte) (charToInt(high) * 16 + charToInt(low));
}
return bytes;
}
/**
*
*
* @param hexString
* @return
*/
public byte[] hexStrToBinaryStr(String hexString) {
if (hexString=="") {
return null;
}
hexString = hexString.replaceAll(" ", "");
int len = hexString.length();
int index = 0;
byte[] bytes = new byte[len / 2];
while (index < len) {
String sub = hexString.substring(index, index + 2);
bytes[index/2] = (byte)Integer.parseInt(sub,16);
index += 2;
}
return bytes;
}
/**
* Compare the both byte array
*
* @param first the first byte array
* @param second the second byte array
* @return if true first equals second ,otherwise unequal
*/
public boolean compareBytes(byte[] first, byte[] second) {
if (first.length != second.length) {
return false;
}
for (int i = 0; i < first.length; i++) {
if (first[i] != second[i]) {
return false;
}
}
return true;
}
/**
* Convert Ascii string to byte array.
*
* @param string the needed converter data.
* @return byte[] the converted value.
*/
public byte[] asciiStringToBytes(String string) {
byte[] result = new byte[string.length()];
for (int i = 0; i < string.length(); i++) {
result[i] = (byte) string.charAt(i);
}
return result;
}
public String computeCRC(String data) {
data = data.replace(" ", "");
int len = data.length();
if (!(len % 2 == 0)) {
return "0000";
}
int num = len / 2;
byte[] para = new byte[num];
for (int i = 0; i < num; i++) {
int value = Integer.valueOf(data.substring(i * 2, 2 * (i + 1)), 16);
para[i] = (byte) value;
}
return getCRC(para);
}
/**
* CRC16
*
* @param bytes
*
* @return {@link String}
* @since 1.0
*/
public String getCRC(byte[] bytes) {
// CRC寄存器全为1
int CRC = 0x0000ffff;
// 多项式校验值
int POLYNOMIAL = 0x0000a001;
int i, j;
for (i = 0; i < bytes.length; i++) {
CRC ^= ((int) bytes[i] & 0x000000ff);
for (j = 0; j < 8; j++) {
if ((CRC & 0x00000001) != 0) {
CRC >>= 1;
CRC ^= POLYNOMIAL;
} else {
CRC >>= 1;
}
}
}
// 结果转换为16进制
String result = Integer.toHexString(CRC).toUpperCase();
if (result.length() != 4) {
StringBuffer sb = new StringBuffer("0000");
result = sb.replace(4 - result.length(), 4, result).toString();
}
//高位在前地位在后
//return result.substring(2, 4) + " " + result.substring(0, 2);
// 交换高低位,低位在前高位在后
return result.substring(2, 4) + " " + result.substring(0, 2);
}
public static int byteToInt(byte b) {
return (b) & 0xff;
}
}

@ -0,0 +1,6 @@
server:
port: 11727
service:
port: 11728
log:
debug: false

@ -0,0 +1,3 @@
spring:
profiles:
active: dev

@ -0,0 +1,49 @@
Configuration:
status: warn
Properties: # 定义全局变量
Property: # 缺省配置用于开发环境。其他环境需要在VM参数中指定如下
#测试:-Dlog.level.console=warn -Dlog.level.xjj=trace
#生产:-Dlog.level.console=warn -Dlog.level.xjj=info
- name: log.level.console
value: trace
- name: log.path
value: ../logs #
- name: project.name
value: netty-all
Appenders:
Console: #输出到控制台
name: CONSOLE
target: SYSTEM_OUT
PatternLayout:
#pattern: "%d{yyyy-MM-dd HH:mm:ss,SSS}:%4p %t (%F:%L) - %m%n"
pattern: "%d [%X{X-B3-TraceId},%X{X-B3-SpanId},%X{X-B3-ParentSpanId},%X{X-Span-Export}] %-5p %c:%L [%t] - %m%n"
RollingFile:
- name: ROLLING_FILE
ignoreExceptions: false
fileName: ${log.path}/${project.name}.log
filePattern: "${log.path}/$${date:yyyy-MM}/${project.name}-%d{yyyy-MM-dd}-%i.log"
PatternLayout:
pattern: "%d [%X{X-B3-TraceId},%X{X-B3-SpanId},%X{X-B3-ParentSpanId},%X{X-Span-Export}] %-5p %c:%L [%t] - %m%n"
Policies:
TimeBasedTriggeringPolicy:
modulate: true
interval: 1
SizeBasedTriggeringPolicy:
size: "50M"
DefaultRolloverStrategy:
max: 100
Loggers:
Root:
level: info
AppenderRef:
- ref: CONSOLE
- ref: ROLLING_FILE
Logger:
- name: com.platform
additivity: false
level: debug
AppenderRef:
- ref: CONSOLE
- ref: ROLLING_FILE
Loading…
Cancel
Save