C++ 连包分包处理

所谓连包, 分包问题. 比如说. 根据通讯协议, 发送了两条命令. 第一条命令 80 字节. 第二条命令 120 字节. 但接受方收到数据的时候, 可能第一次收到 150 个字节. 第二次收到 50 个字节. 要正确理解发送方的命令包. 接受方必须根据协议正确的从第一包数据中取出前 80 个字节. 作为一条命令, 再把剩下的 70 个字节和第二包的数据组合成第二条命令. 这个问题如此常见. 以致于实现一个可以复用的, 轻量级的解析器就变得很有必要. 于是就有了这个 PacketExtractor 类. PacketExtractor 用 C++ 实现, 运用分层的思想. 结合模板, 虚函数等技术. 让使用者只需要重载几个函数, 就可以完美的解决在通讯过程中的拆包, 组包的问题.

设计思路
PacketExtractor 的算法其实很朴素.

在收到一包数据之后, 先查找包头特征字(比如 7e), 一般的, 协议中都有一个包长字段. 将包头作为包的起始字节, 找到包长. 如果收到的包还不够包长指定的长度, 则将包放到一个缓冲区. 等下一包数据到的时候, 将这次的半包附加到收到数据前面. 递归调用处理函数. 如果收到的数据已经够一包的长度了. 则根据协议取出校验字段. 对包进行校验. 如果通过. 则证明是一个合法包. 交给上层处理. 如果校验失败. 则可能是传输中出现查错. 更多的情况是找到的包头(7e) 并不是真正的包头. 也许是包中间的数据. 这个时候, 在这个 7e 之后继续查找包头. 找到之后继续之前的步骤.
实现细节
有了设计思路， 具体实现就简单了。以下是类的主要成员函数以及变量：

template <DWORD MAX_PACKET, DWORD MIN_PACKET>

class PacketExtractor

{

// 核心函数. 从底层接受数据, 拆包组包, 校验合法之后交给上层处理

virtual void OnRawDataReceived(const BYTE byBuf[], DWORD dwLen);

// return zero-based position of header in the packet .

// 如果没有找到包头, 那么返回 dwLen 的值

virtual UINT LocatePacketHeader(const BYTE byBuf[], DWORD dwLen);

// 由收到的包获取这个包的协议包长.

virtual UINT ExtractPacketLength(const BYTE byBuf[], DWORD dwLen);

// 是否一个有效的包.可以在派生类中定义

virtual BOOL IsPacketValid(const BYTE [], DWORD);

/* 收到一个合法有效的包时调用这个函数, 交给上层处理 */

virtual void OnPacketReceived(const BYTE byBuf[], DWORD dwLen) = 0;

}

其中, OnRawDataReceived() 函数实现了流程图所示的算法. 另外几个函数, 考虑到代码重用, 定义为虚函数.

LocatePacketHeader()

重载此函数来定位包头所在的位置. 根据协议的不同, 查找协议特定的包头特征字. 返回包头在缓冲区中的位置. 如果没有找到包头. 则返回 dwLen.

ExtractPacketLength()

PacketExtractor 找到包头, 并且确保收到足够多的数据之后, 调用此函数来确定此包数据的协议包长. 这里的协议包长是指协议中的包长字段指示的长度. PacketExtractor 用这个长度确定一包命令的结束.

IsPacketValid()

PacketExtractor 从包头开始算起, 取出协议包长指定长度的数据. 传给 IsPacketValid(), 用户可以重载此函数, 在其中根据协议定义的校验方法验证这个命令包是否正确. 如果是有效包, 返回 TRUE, 否则返回 FALSE.

OnPacketReceived()

PacketExtractor 通过这个函数将有效的数据包交给上层处理. 用户可重载这个函数, 实现自己的业务逻辑.

用户从 PacketExtractor 派生自己的类. 然后重载这四个虚函数. 就可以在 OnPacketReceived() 收到合法包用以处理业务逻辑. 另外, 在 PacketExtractor 的处理过程中, 需要知道协议定义的最大包长和最小包长. 最大包长用来定义缓冲区. 最小包长用来确定是否已经收到足够的数据以调用 ExtractPacketLength() 函数. 由于这两个值在编译时即可确定. 所以, 用模板参数来指定.

具体应用
举一个简单的例子. 比如协议的定义如下:

7E 07 02 FF FF FF 84

其中 7E 为包头, 07 为包长. 02 FF FF FF 为数据段. 数据段长度最小为 0, 最大为 100. 最后一个字节 84 为累加校验和. 这个例子只是为了表意. 实际的协议可能采用两个或者更多字节作为包头, 减少数据段中的冲突(如果要彻底解决冲突问题, 可以采用转义字符. 比如将除了包头之外的 7e 都转义成其他字符). 校验可能会采用 CRC. 更加可靠.

要解析这样一个协议, 可以定义下面的类.

// MyPacketExtractor.h header file

#include “PacketExtractor.h”

class MyPacketExtractor : public PacketExtractor<100 + 3/*最大数据段长度 + 包头 + 包长字段 + 校验字段*/, 3 /*数据段最小为 0 */>

{

virtual UINT LocatePacketHeader(const BYTE byBuf[], DWORD dwLen)

{

for (int i = 0; i < dwLen; ++ i)

if (byBuf[i] == 0x7e) return i;

return dwLen;

}

virtual UINT ExtractPacketLength(const BYTE byBuf[], DWORD){return byBuf[1]; /* 包长位于数据包的第二个字节. 直接返回之*/}

virtual BOOL IsPacketValid(const BYTE [], DWORD dwLen)

{

BYTE byChksum = 0;

for (int i = 0; i < dwLen - 1; ++ i)

byChksum += byBuf[i];

return byChksum == byBuf[dwLen - 1];

}

virtual void OnPacketReceived(const BYTE byBuf[], DWORD dwLen)

{

// 在这里处理业务逻辑

printf("recv a packet. len = %d\n", dwLen);

}

}

下面的代码演示了如何使用 MyPacketExtractor. 代码基于MFC, 使用CAsyncSocket通讯类.

// asyncsock.cpp

// compiled in vc6. cmdline: cl.exe -GX -MTd asyncsock.cpp

#include <afxwin.h> // MFC core and standard components

#include <afxsock.h> // MFC socket extensions

#include <stdio.h>

#include "MyPacketExtractor.h"

const int BUF_LEN = 1024;

class CDataSocket : public CAsyncSocket, public MyPacketExtractor {

void OnReceive(int) {

int nRecv = Receive(byBuf, BUF_LEN);

do OnRawDataReceived(byBuf, nRecv);

while ((nRecv = Receive(byBuf, BUF_LEN)) == BUF_LEN);

}

BYTE byBuf[BUF_LEN + 1];

void OnClose(int nErr) {

PostQuitMessage(nErr);

}

};

class CListenSocket : public CAsyncSocket {

void OnAccept(int nErr) {

m_datasock.Close();

printf("OnAccept\n", nErr);

Accept(m_datasock);

m_datasock.AsyncSelect(FD_READ | FD_CLOSE);

}

CDataSocket m_datasock;

};

void main(int argc, char** argv) {

bool bServer = !(argc<=1);

AfxWinInit(::GetModuleHandle(NULL), NULL, NULL, 0);

AfxSocketInit();

CListenSocket sock;

CDataSocket sock2;

CAsyncSocket* psock = bServer ? (CAsyncSocket*)&sock : &sock2;

if (bServer) {

psock->Create(5000);

psock->AsyncSelect(FD_ACCEPT);

psock->Listen();

} else {

psock->Create();

psock->Connect("127.0.0.1", 5000);

}

while (1) {

MSG msg;

while (PeekMessage(&msg, 0, 0, 0, PM_NOREMOVE)) {

if (GetMessage(&msg, 0, 0, 0))

DispatchMessage(&msg);

else exit(msg.wParam);

}

Sleep(100);

if (! bServer) {

static BYTE byFactor = 0, byBuf[] = {

0x7e, 1,5,1,3,5,6,3,1,6,8, // 故意的 7e 开头, 非法数据包.

0x7E, 07, 02, 0xFF, 0xFF, 0xFF, 0x84, // 有效包 1

0x1, 2, 3, 4, 5, 6, // 垃圾数据

0x7e, 0x0d, 03, 0x7e, 0x13, 0x52, 0x7e, 0x7e, 0x33, 0x14, 0x85, 0x64, 0x9d,// 有效包 2. 但中间故意夹杂了 7e

54,65,68,45,32,12, // 垃圾数据

0x7E, 0x8, 05, 0xFF, 0xFF, 0x0F, 0xf0, 0x88 // 有效包 3.

};

if (! (byFactor ++ % 10))

psock->Send(byBuf, sizeof(byBuf));

}

}

}

这个例子将服务器, 客户端写到了一起. 根据命令行参数区分. 如果没有命令行参数, 则作为客户端, 否则作为服务端. 先运行服务端程序, 开始监听, 然后运行客户端. 连接服务器. 建立连接之后, 每秒发送一包数据. 这包数据中包含了 3 个有效包, 其间还夹杂了一些无效包和无效数据. 从运行结果可以看出, 服务端收到数据之后正确的解析了其中的有效包. 并把有效包的包长打印出来. 程序运行截图如下:

需要注意的是. 有的协议可能没有包长字段. 而是采用一个包尾字段来作为包的结束. 则 ExtractPacketLength() 可以这么写:

UINT ExtractPacketLength(const BYTE byBuf[], DWORD dwLen)

{

for (int i = 0; i < dwLen; ++ i)

{

if (byBuf[i] == PACKET_TAIL)

return i + 1;

}

return dwLen + 1;

}

如果找到了包尾, 那自然可以得出包长, 如果没有找到. 则返回一个比当前收到的数据包的长度还大的一个值. 告诉 PacketExtractor 收到的数据还不够组成一个合法包. PacketExtractor 知道了这是一个半包, 会在下次收到数据的时候将收到的数据附加到这个半包之后继续处理.

结语