澧县网站设计,乐清网站建设,好用的网站,石湾网站建设1、背景
tcp传输的时候会自动拆包#xff0c;因此服务端接收的数据段可能跟客户端发送过来的数据段长度不一致#xff0c;比如客户端一次发送10000个字节。但是服务端接收了两次才接收完整#xff08;例如第一次接收6000字节#xff0c;第二次接收4000字节#xff09;。但…1、背景
tcp传输的时候会自动拆包因此服务端接收的数据段可能跟客户端发送过来的数据段长度不一致比如客户端一次发送10000个字节。但是服务端接收了两次才接收完整例如第一次接收6000字节第二次接收4000字节。但是服务端每次必须要接收完所有的字节才能进行处理而且客户端每次发的数据长度都不一致。 于是经过协商客户端每次发送数据段时在数据段前加10个字节后面统一称数据包头前6个字节为数据包起始标识符后4个字节为此次发送数据段的长度。
2、难点
因为tcp会拆包所以数据段前的10个字节可能会出现在任何位置也可能会出现在两次tcp传输过程中。另外如果包头前6个字节不是指定的标识要向后顺延直到找出包头。
3、思路
1使用两个ByteBuffer对象一个记录数据段前的10个字节该对象仅创建一次。另一个ByteBuffer对象存储去除包头后的完整的数据段信息该对象在每次接收新的包头时都会根据包头的后4个字节重新创建因为jvm的自动垃圾回收所以这里不用担心内存溢出问题。
2接收完整的数据段后如果还有多余数据则使用迭代方式处理。
4、java代码实现
1、这里只列出了核心代码相关逻辑需要自己补全2、创建tcp服务端代码
try (ServerSocket ss new ServerSocket(port)) {while (true) {Socket socket ss.accept();new SocketHandler(socket, eqpmtId, port, save).start();}
} catch (Exception e) {log.error(TCP服务端创建异常端口为{}异常为\n, this.port, e);
}3、tcp服务端详细处理代码
Slf4j
class SocketHandler extends Thread {private Socket socket;private String eqpmtId;private Integer port;private boolean save;public SocketHandler(Socket socket, String eqpmtId, Integer port, boolean save) {this.socket socket;this.eqpmtId eqpmtId;this.port port;this.save save;}Overridepublic void run() {log.info(与{},{}建立消息socket通信, eqpmtId, port);try (InputStream inputStream socket.getInputStream();FileOutputStream os new FileOutputStream(new File(D:\\tmp-data\\ System.currentTimeMillis() .h264));) {byte[] buffer new byte[64 * 1024];int len 0;ByteBuffer dataBuffer null;ByteBuffer headBuffer ByteBuffer.allocate(10);while (socket.isConnected() !socket.isClosed()) {if ((len inputStream.read(buffer)) ! -1) {log.info(收到数据包len{}, len);try {dataBuffer getDataBuffer(buffer, 0, len, headBuffer, dataBuffer);} catch (Exception e) {log.error(接收数据异常重新开始接收...\n,e);headBuffer.clear();dataBuffer.clear();}} else {log.info(没有数据休眠1秒否则cpu会飙升);Thread.sleep(1000);}}} catch (Exception e) {log.error(socket传输异常异常为\n, this.port, e);}log.info(关闭与},{}消息socket通信, eqpmtId, port);}private ByteBuffer getDataBuffer(byte[] buffer, int start, int end, ByteBuffer headBuffer, ByteBuffer dataBuffer) {int offset start;int tmpLen 0;//先找到包头if (headBuffer.position() headBuffer.capacity()) {//当前数组长小于包头长度有整个数组放入头缓存后返回int len end - offset;if (len headBuffer.capacity() - headBuffer.position()) {headBuffer.put(buffer, offset, len);return dataBuffer;}tmpLen headBuffer.capacity() - headBuffer.position();headBuffer.put(buffer, offset, headBuffer.capacity() - headBuffer.position());offset offset tmpLen;//包头缓存填充满了判断包头是否正确if (!isHead(headBuffer.array())) {//包头不正确则不断向后移位直到找到包头log.info(包头有问题向后移动一位继续校验);int headLastIndex headBuffer.capacity() - 1;for (; offset end; offset) {for (int i 0; i headLastIndex; i) headBuffer.put(i, headBuffer.get(i 1));headBuffer.put(headLastIndex, buffer[offset]);if (isHead(headBuffer.array())) break;}//移位结束确认是找到了包头还是当前数组已经遍历完if (!isHead(headBuffer.array())) {headBuffer.position(headLastIndex);return dataBuffer;}}//包头正确后解析获取数据包有多长并创建对应的缓存对象int dataLen dataLen(headBuffer.array());log.info(包头设定长度为{}, dataLen);dataBuffer ByteBuffer.allocate(dataLen);}if (offset end) return dataBuffer;//如果可以填充满数据缓存对象则发送数据包并清理缓存if (end - offset dataBuffer.capacity() - dataBuffer.position()) {tmpLen dataBuffer.capacity() - dataBuffer.position();dataBuffer.put(buffer, offset, dataBuffer.capacity() - dataBuffer.position());offset offset tmpLen;/** 收到完整数据包进行处理注意这里的函数要替换成自己的处理逻辑 **/sendData(dataBuffer, null);dataBuffer.clear();headBuffer.clear();if (offset end) return dataBuffer;//迭代处理剩下的数据return getDataBuffer(buffer, offset, end, headBuffer, dataBuffer);}//如果不能填充慢数据缓存对象则整个数据放入后返回dataBuffer.put(buffer, offset, end - offset);return dataBuffer;}//判断是否为包头public boolean isHead(byte[] buffer) {if (buffer null || buffer.length 10) return false;int b1 buffer[0];int b2 buffer[1];int b3 buffer[2];int b4 buffer[3];int b5 buffer[4];int b6 buffer[5];String s b1 b2 b3 b4 b5 b6;if (001001.equals(s)) return true;return false;}//判断数据包的长度ByteUtil用的hutool工具包里的类也可以自己实现public int dataLen(byte[] buffer) {return ByteUtil.bytesToInt(new byte[]{buffer[6], buffer[7], buffer[8], buffer[9]});}}
