1、背景

tcp傳輸?shù)臅r候會自動拆包，因此服務(wù)端接收的數(shù)據(jù)段可能跟客戶端發(fā)送過來的數(shù)據(jù)段長度不一致，比如客戶端一次發(fā)送10000個字節(jié)。但是服務(wù)端接收了兩次才接收完整（例如第一次接收6000字節(jié)，第二次接收4000字節(jié)）。但是服務(wù)端每次必須要接收完所有的字節(jié)才能進(jìn)行處理，而且客戶端每次發(fā)的數(shù)據(jù)長度都不一致。
于是經(jīng)過協(xié)商，客戶端每次發(fā)送數(shù)據(jù)段時，在數(shù)據(jù)段前加10個字節(jié)（后面統(tǒng)一稱數(shù)據(jù)包頭），前6個字節(jié)為數(shù)據(jù)包起始標(biāo)識符，后4個字節(jié)為此次發(fā)送數(shù)據(jù)段的長度。

2、難點(diǎn)

因?yàn)閠cp會拆包，所以數(shù)據(jù)段前的10個字節(jié)可能會出現(xiàn)在任何位置，也可能會出現(xiàn)在兩次tcp傳輸過程中。另外如果包頭前6個字節(jié)不是指定的標(biāo)識，要向后順延，直到找出包頭。

3、思路

1）使用兩個ByteBuffer對象，一個記錄數(shù)據(jù)段前的10個字節(jié)，該對象僅創(chuàng)建一次。另一個ByteBuffer對象存儲去除包頭后的完整的數(shù)據(jù)段信息，該對象在每次接收新的包頭時，都會根據(jù)包頭的后4個字節(jié)重新創(chuàng)建（因?yàn)閖vm的自動垃圾回收，所以這里不用擔(dān)心內(nèi)存溢出問題）。

2）接收完整的數(shù)據(jù)段后，如果還有多余數(shù)據(jù)則使用迭代方式處理。

4、java代碼實(shí)現(xiàn)

1、這里只列出了核心代碼，相關(guān)邏輯需要自己補(bǔ)全2、創(chuàng)建tcp服務(wù)端代碼
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服務(wù)端創(chuàng)建異常，端口為{}，異常為\n", this.port, e);
}3、tcp服務(wù)端詳細(xì)處理代碼
@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("收到數(shù)據(jù)包len={}", len);try {dataBuffer = getDataBuffer(buffer, 0, len, headBuffer, dataBuffer);} catch (Exception e) {log.error("接收數(shù)據(jù)異常，重新開始接收...\n",e);headBuffer.clear();dataBuffer.clear();}} else {log.info("沒有數(shù)據(jù)，休眠1秒，否則cpu會飆升");Thread.sleep(1000);}}} catch (Exception e) {log.error("socket傳輸異常，異常為\n", this.port, e);}log.info("關(guān)閉與},{}消息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()) {//當(dāng)前數(shù)組長小于包頭長度有，整個數(shù)組放入頭緩存后返回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("包頭有問題，向后移動一位繼續(xù)校驗(yàn)");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;}//移位結(jié)束確認(rèn)是找到了包頭還是當(dāng)前數(shù)組已經(jīng)遍歷完if (!isHead(headBuffer.array())) {headBuffer.position(headLastIndex);return dataBuffer;}}//包頭正確后，解析獲取數(shù)據(jù)包有多長，并創(chuàng)建對應(yīng)的緩存對象int dataLen = dataLen(headBuffer.array());log.info("包頭設(shè)定長度為{}", dataLen);dataBuffer = ByteBuffer.allocate(dataLen);}if (offset == end) return dataBuffer;//如果可以填充滿數(shù)據(jù)緩存對象，則發(fā)送數(shù)據(jù)包，并清理緩存if (end - offset >= dataBuffer.capacity() - dataBuffer.position()) {tmpLen = dataBuffer.capacity() - dataBuffer.position();dataBuffer.put(buffer, offset, dataBuffer.capacity() - dataBuffer.position());offset = offset + tmpLen;/** 收到完整數(shù)據(jù)包，進(jìn)行處理，注意這里的函數(shù)要替換成自己的處理邏輯 **/sendData(dataBuffer, null);dataBuffer.clear();headBuffer.clear();if (offset == end) return dataBuffer;//迭代處理剩下的數(shù)據(jù)return getDataBuffer(buffer, offset, end, headBuffer, dataBuffer);}//如果不能填充慢數(shù)據(jù)緩存對象，則整個數(shù)據(jù)放入后返回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;}//判斷數(shù)據(jù)包的長度（ByteUtil用的hutool工具包里的類，也可以自己實(shí)現(xiàn)）public int dataLen(byte[] buffer) {return ByteUtil.bytesToInt(new byte[]{buffer[6], buffer[7], buffer[8], buffer[9]});}}