blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/JfptApplication.java
@@ -19,6 +19,7 @@ import org.springblade.core.cloud.feign.EnableBladeFeign; import org.springblade.core.launch.BladeApplication; import org.springblade.jfpt.nettyServer.Server; import org.springblade.jfpt.nettyTcpServer.TcpServer; import org.springblade.jfpt.nettyUdpServer.server.UdpServer; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -39,8 +40,9 @@ @Override public void run(String... args) throws Exception { Server server=new Server(8088); UdpServer udpServer=new UdpServer(8099); //Server server=new Server(8088); TcpServer server=new TcpServer(8088); //UdpServer udpServer=new UdpServer(8099); } } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyTcpServer/MyNettyDecode.java
New file @@ -0,0 +1,126 @@ package org.springblade.jfpt.nettyTcpServer; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import java.util.List; /** * 自定义解码器 * */ public class MyNettyDecode extends ByteToMessageDecoder { public final int BASE_LENGTH = 4 + 4 ; // @Override // protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception { // System.out.println("byteBuf = " + byteBuf.readableBytes()); // // 可读长度必须大于基本长度 // if (byteBuf.readableBytes() >= BASE_LENGTH) { // // 防止socket字节流攻击 // // 防止，客户端传来的数据过大 // // 因为，太大的数据，是不合理的 // if (byteBuf.readableBytes() > 202048) { // byteBuf.skipBytes(byteBuf.readableBytes()); // } // // // 记录包头开始的index // int beginReader; // // while (true) { // // 获取包头开始的index // beginReader = byteBuf.readerIndex(); // // 标记包头开始的index // byteBuf.markReaderIndex(); // // 读到了协议的开始标志，结束while循环 // if (byteBuf.readInt() == NettyConstant.HEAD_DATA) { // break; // } // // // 未读到包头，略过一个字节 // // 每次略过，一个字节，去读取，包头信息的开始标记 // byteBuf.resetReaderIndex(); // byteBuf.readByte(); // // // 当略过，一个字节之后， // // 数据包的长度，又变得不满足 // // 此时，应该结束。等待后面的数据到达 // if (byteBuf.readableBytes() < BASE_LENGTH) { // return; // } // } // // // 消息的长度 // // int length = byteBuf.readInt(); // // 判断请求数据包数据是否到齐 // if (byteBuf.readableBytes() < length) { // // 还原读指针 // byteBuf.readerIndex(beginReader); // return; // } // // // 读取data数据 // byte[] data = new byte[length]; // byteBuf.readBytes(data); // System.out.println("byteBuf末尾 = " + byteBuf); // // NettyProPool protocol = new NettyProPool(data.length, data); // list.add(protocol); // System.out.println("list = " + list); // } // } static final int PACKET_SIZE = 110088; // 用来临时保留没有处理过的请求报文 ByteBuf tempMsg = Unpooled.buffer(); /** * @param ctx * @param in 请求的数据 * @param out 将粘在一起的报文拆分后的结果保留起来 * @throws Exception */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { System.out.println(Thread.currentThread() + "收到了一次数据包，长度是：" + in.readableBytes()); // 合并报文 ByteBuf message = null; int tmpMsgSize = tempMsg.readableBytes(); System.out.println("tmpMsgSize = " + tmpMsgSize); // 如果暂存有上一次余下的请求报文，则合并 if (tmpMsgSize > 0) { message = Unpooled.buffer(); message.writeBytes(tempMsg); message.writeBytes(in); System.out.println("合并：上一数据包余下的长度为：" + tmpMsgSize + ",合并后长度为:" + message.readableBytes()); } else { message = in; } int size = message.readableBytes(); int counter = size / PACKET_SIZE; for (int i = 0; i < counter; i++) { byte[] request = new byte[PACKET_SIZE]; // 每次从总的消息中读取220个字节的数据 message.readBytes(request); // 将拆分后的结果放入out列表中，交由后面的业务逻辑去处理 out.add(Unpooled.copiedBuffer(request)); } // 多余的报文存起来 // 第一个报文： i+ 暂存 // 第二个报文： 1 与第一次 size = message.readableBytes(); if (size != 0) { System.out.println("多余的数据长度：" + size); // 剩下来的数据放到tempMsg暂存 tempMsg.clear(); tempMsg.writeBytes(message.readBytes(size)); } } } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyTcpServer/NettyConstant.java
New file @@ -0,0 +1,12 @@ package org.springblade.jfpt.nettyTcpServer; /** * 自定义协议常量 */ public class NettyConstant { /** * 自定义协议开始 */ public static final int HEAD_DATA = 0*76; } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyTcpServer/NettyProPool.java
New file @@ -0,0 +1,53 @@ package org.springblade.jfpt.nettyTcpServer; public class NettyProPool { /** * 协议开始的标志，16进制为 0*76 */ private int haed_data=NettyConstant.HEAD_DATA; /** * 传送的长度 */ private int contentLength; /** * 传送的内容 */ private byte[] content; public NettyProPool(int contentLength) { this.contentLength = contentLength; } public int getContentLength() { return contentLength; } public void setContentLength(int contentLength) { this.contentLength = contentLength; } public NettyProPool(int contentLength, byte[] content) { this.contentLength = contentLength; this.content = content; } public byte[] getContent() { return content; } public void setContent(byte[] content) { this.content = content; } public int getHaed_data() { return haed_data; } public void setHaed_data(int haed_data) { this.haed_data = haed_data; } } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyTcpServer/ParseString.java
New file @@ -0,0 +1,14 @@ package org.springblade.jfpt.nettyTcpServer; import io.swagger.models.auth.In; public class ParseString { public static Integer[] stringToInts(String s){ String[] split = s.split(","); Integer[] n = new Integer[split.length]; for(int i = 0;i<split.length;i++){ n[i] = Integer.parseInt(split[i]); } return n; } } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyTcpServer/TcpServer.java
New file @@ -0,0 +1,92 @@ package org.springblade.jfpt.nettyTcpServer; import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.ServerSocketChannel; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; import io.netty.handler.codec.LengthFieldPrepender; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.logging.LoggingHandler; import java.nio.ByteOrder; import java.nio.charset.Charset; public class TcpServer { private int port; private ServerSocketChannel serverSocketChannel; public TcpServer(int port){ this.port = port; bind(); } private void bind() { Thread thread = new Thread(new Runnable() { @Override public void run() { //服务端要建立两个group，一个负责接收客户端的连接，一个负责处理数据传输 //连接处理group EventLoopGroup boss = new NioEventLoopGroup(); //事件处理group EventLoopGroup worker = new NioEventLoopGroup(); ServerBootstrap bootstrap = new ServerBootstrap(); // 绑定处理group bootstrap.group(boss, worker).channel(NioServerSocketChannel.class) //保持连接数 .option(ChannelOption.SO_BACKLOG, 128) //有数据立即发送 .option(ChannelOption.TCP_NODELAY, true) //保持连接 .childOption(ChannelOption.SO_KEEPALIVE, true) .childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(512*250)) .childHandler(new NettyChannelHandler()); //绑定端口，同步等待成功 ChannelFuture future; try { future = bootstrap.bind(port).sync(); if (future.isSuccess()) { serverSocketChannel = (ServerSocketChannel) future.channel(); System.out.println("服务端启动成功，端口："+port); } else { System.out.println("服务端启动失败！"); } //等待服务监听端口关闭,就是由于这里会将线程阻塞，导致无法发送信息，所以我这里开了线程 future.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { //优雅地退出，释放线程池资源 boss.shutdownGracefully(); worker.shutdownGracefully(); } } }); thread.start(); } private class NettyChannelHandler extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { //socketChannel.pipeline().addLast(new MyNettyDecode()); //socketChannel.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8"))); //socketChannel.pipeline().addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4)); //socketChannel.pipeline().addLast("encoder", new LengthFieldPrepender(4, false)); socketChannel.pipeline().addLast(new TcpServerHandler()); } } } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyTcpServer/TcpServerHandler.java
New file @@ -0,0 +1,105 @@ package org.springblade.jfpt.nettyTcpServer; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import org.springblade.jfpt.alarm.service.IAlarmService; import org.springblade.jfpt.equipment.service.IEquipmentService; import org.springblade.jfpt.nettyServer.NettyConfig; import org.springblade.jfpt.nettyUdpServer.service.AnimalHeatService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.util.concurrent.ConcurrentHashMap; @Component public class TcpServerHandler extends ChannelInboundHandlerAdapter { private String reg_LA ="LA[d]{8}[d|A-F]{12}[d|A-F]{8}[d|A-Z]{2}[d|A-F]{4}[x2A][d|A-F]{6}[#@]"; private String reg_LB = "LB[\\d|A-F]{12}[\\x2A].*[#@]"; private String reg_LB2 = "LB[\\d|A-F]{6}[\\x2A].*[#@]"; private String reg_LD = "LD[d]{8}[d|A-F]{12}:[A-Z]{4}[\\x2A].*[#@]"; private ConcurrentHashMap<String, Channel> sessionChannelMap = new ConcurrentHashMap<String, Channel>(); @Autowired private org.springblade.jfpt.catalog.service.catalogService catalogService; @Autowired private org.springblade.jfpt.Netty.service.nettyService nettyService; @Autowired private IEquipmentService equipmentService; @Autowired private IAlarmService alarmService; private static TcpServerHandler tcpServerHandler; @Autowired private AnimalHeatService animalHeatService; @PostConstruct public void init() { tcpServerHandler = this; } /** * 客户端与服务端创建连接的时候调用 */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("CTX:"+ctx.channel()); System.out.println("客户端与服务端连接开始..."); // ChannelId id = ctx.channel().id(); // String s = id.toString(); //NettyConfig.group.add(ctx.channel()); } /** * 客户端与服务端断开连接时调用 */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println("客户端与服务端连接关闭..."); NettyConfig.group.remove(ctx.channel()); } /** * 服务端接收客户端发送过来的数据结束之后调用 */ @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); System.out.println("信息接收完毕..."); } /** * 工程出现异常的时候调用 */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } /** * 服务端处理客户端websocket请求的核心方法，这里接收了客户端发来的信息 */ @Override public void channelRead(ChannelHandlerContext channelHandlerContext, Object info) throws Exception { System.out.println("接收到了："+info); ByteBuf buf = (ByteBuf) info; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "UTF-8"); String content = body; System.out.println("接收客户端数据:" + body); //插入数据 tcpServerHandler.animalHeatService.save(body); } } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyUdpServer/entity/BladeAnimalHeat.java
@@ -37,10 +37,10 @@ private Float tempRed;// float(4,2) DEFAULT NULL COMMENT '当前人员的温度', @TableField("person_rect") private String personRect;// varchar(63) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '人脸位置二位数组，[left,top,right,bottom]', private String[] personRect;// varchar(63) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '人脸位置二位数组，[left,top,right,bottom]', @TableField("person_rect_red") private String personRectRed;// varchar(63) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '红外人脸位置二位数组，[left,top,right,bottom]', private Integer[] personRectRed;// varchar(63) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '红外人脸位置二位数组，[left,top,right,bottom]', @TableField("create_time") private Date createTime;// datetime DEFAULT NULL COMMENT '创建时间', blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyUdpServer/service/AnimalHeatService.java
@@ -1,7 +1,15 @@ package org.springblade.jfpt.nettyUdpServer.service; import com.baomidou.mybatisplus.extension.service.IService; import org.springblade.jfpt.nettyUdpServer.entity.BladeAnimalHeat; /** * 体温监测服务层接口层 */ public interface AnimalHeatService { public interface AnimalHeatService extends IService<BladeAnimalHeat> { /** * 插入体温数据 * @param body 温度数据 */ void save(String body); } blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyUdpServer/service/impl/AnimalHeatServiceImpl.java
@@ -1,12 +1,22 @@ package org.springblade.jfpt.nettyUdpServer.service.impl; import com.alibaba.fastjson.JSONObject; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import org.springblade.jfpt.nettyTcpServer.ParseString; import org.springblade.jfpt.nettyUdpServer.entity.BladeAnimalHeat; import org.springblade.jfpt.nettyUdpServer.mapper.AnimalHeatMapper; import org.springblade.jfpt.nettyUdpServer.service.AnimalHeatService; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.lang.reflect.Array; import java.util.Map; import static org.springblade.core.tool.utils.DateUtil.now; /** * 体温数据服务层 * */ @Service public class AnimalHeatServiceImpl extends ServiceImpl<AnimalHeatMapper, BladeAnimalHeat> implements AnimalHeatService { @Override @@ -14,4 +24,33 @@ return false; } /** * 新增体温数据 * @param body 体温数据 */ @Override @Transactional public void save(String body) { //1.截取信息 int i = body.indexOf("{"); //2.从 ｛ 处开始截取 int length = body.length(); //3.截取 String dataBody = body.substring(i, length); System.out.println("dataBody = " + dataBody); //4.数据转换 Map<String,Object> data = (Map<String, Object>) JSONObject.parse(dataBody); System.out.println("data = " + data); //5.创建体温数据对象 BladeAnimalHeat bladeAnimalHeat = new BladeAnimalHeat(); //6.取出数据，封装数据到对象 bladeAnimalHeat.setPicture(data.get("picture").toString()); String temp_red =data.get("temp_red").toString(); bladeAnimalHeat.setTempRed(Float.parseFloat(temp_red.substring(1,temp_red.length()-1))); String person_rects = data.get("person_rect").toString(); bladeAnimalHeat.setPersonRect(person_rects.substring(2,person_rects.length()-2).split(",")); bladeAnimalHeat.setCreateTime(now()); //7.插入数据 int idata = baseMapper.insert(bladeAnimalHeat); } }
