C++解决TCP粘包

TCP粘包问题

TCP是面向连接的，面向流的可靠性传输。TCP会将多个间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包发送，这样一个数据包里就可能含有多个消息的数据，面向流的通信是无消息保护边界的，也就是TCP粘包。接收端需要自己完成数据的拆包和组包，解决粘包问题。

要解决TCP粘包问题，就要给TCP定义公共包头，包头一般包括消息类型和消息大小，用包头来分割每个数据包，做数据包的边界。

下面分别用C++实现TCP客户端和TCP服务端，使用qt测试。

TCP客户端

TCP客户端主动连接到TCP服务端，并接收TCP服务端发送的数据，对接收的数据按照定义的公共包头进行分割组包，每当组成一个完整数据包时，打印相关信息。
TcpClient.h

#ifndef TCPCLIENT_H
#define TCPCLIENT_H#include <string.h>
#include <stdint.h>
#include <stdint.h>
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <new>#define MAX_PKT_SIZE        (256<<20)   //网络包最大长度
//业务包头
struct CommMsgHdr
{uint16_t uMsgType;uint32_t uTotalLen;
};typedef struct _TcpHandle_{int32_t fd;uint32_t     uRcvLen;        //已接收数据大小uint32_t     uAllLen;        //消息总长度struct sockaddr_in local_addr;struct sockaddr_in remote_addr;_TcpHandle_(){uRcvLen = 0;uAllLen = 0;}
}TcpHandle;class TcpClient
{
public:TcpClient();int32_t create_tcpClient(char *serverIp, int32_t serverPort);int32_t SendData(char *data, int32_t len);bool m_runing;int epoll_fd;TcpHandle* pTcpHandle;
private:pthread_t threadId;
};#endif // TCPCLIENT_H

TcpClient.cpp

#include "TcpClient.h"int32_t TcpRcv(const int32_t& fd, void* buff, const uint32_t& len)
{int32_t iCurrRecv = recv(fd, buff, len, MSG_NOSIGNAL);if (0 < iCurrRecv) {return iCurrRecv;} else if (iCurrRecv < 0) {if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) {return 0;} else return -1;} else return -1;
}void* DealTcpThread(void* obj)
{TcpClient* pTcpClient = (TcpClient*)obj;TcpHandle* pTcpHandle = pTcpClient->pTcpHandle;const int kEpollDefaultWait = 1;//超时时长，单位msstruct epoll_event alive_events[256];uint32_t recv_buffer_max = 1024 * 1024;uint8_t *recv_buffer = nullptr;recv_buffer = new uint8_t[recv_buffer_max];uint32_t head_len = (uint32_t)sizeof(CommMsgHdr);while (pTcpClient->m_runing){int num = epoll_wait(pTcpClient->epoll_fd, alive_events, 256, kEpollDefaultWait);for (int i = 0; i < num; ++i){int fd = alive_events[i].data.fd;int events = alive_events[i].events;if ( events & EPOLLIN ){//1.开始接收头部if(pTcpHandle->uRcvLen < head_len){int32_t iRecvLen = TcpRcv(fd, recv_buffer + pTcpHandle->uRcvLen, head_len - pTcpHandle->uRcvLen);if (0 == iRecvLen) continue;else if (0 > iRecvLen) {printf("Recv head data, return [%d] and err[%s],fd=[%d].", iRecvLen, strerror(errno),fd);close(fd);//关闭socketcontinue;}pTcpHandle->uRcvLen += iRecvLen;//如果已经接收完整头部if(pTcpHandle->uRcvLen >= head_len){CommMsgHdr* pHdr = (CommMsgHdr *)recv_buffer;pTcpHandle->uAllLen = pHdr->uTotalLen;//如果报文头里的uTotalLen太小或太大，异常处理if ( pHdr->uTotalLen < head_len || pHdr->uTotalLen > MAX_PKT_SIZE ){printf("uTotalLen invalid,uTotalLen=%u,fd=[%d]",pHdr->uTotalLen,fd);close(fd);//关闭socketcontinue;}//如果uTotalLen大于已分配的缓存，重新分配if (((CommMsgHdr *)recv_buffer)->uTotalLen > recv_buffer_max){uint8_t *new_recv_buffer = new uint8_t[((CommMsgHdr *)recv_buffer)->uTotalLen];memcpy(new_recv_buffer, recv_buffer,head_len);delete [] recv_buffer;// 释放原有空间recv_buffer = new_recv_buffer;// 重新指向新开辟的空间recv_buffer_max = ((CommMsgHdr *)recv_buffer)->uTotalLen;// 重新赋值最大buffer长度}}}//2.开始接收数据体else{int32_t iRecvLen = TcpRcv(fd, recv_buffer + pTcpHandle->uRcvLen, pTcpHandle->uAllLen - pTcpHandle->uRcvLen);if (0 == iRecvLen) continue;else if (0 > iRecvLen) {printf("Recv body data, return [%d] and err[%s],fd=[%d].", iRecvLen, strerror(errno),fd);close(fd);//关闭socketcontinue;}pTcpHandle->uRcvLen += iRecvLen;//完成接收if(pTcpHandle->uRcvLen == pTcpHandle->uAllLen){CommMsgHdr* pHdr = (CommMsgHdr*)recv_buffer;printf("Rcv completed,msgType=%d,uTotalLen=%u\n",pHdr->uMsgType,pHdr->uTotalLen);pTcpHandle->uRcvLen = 0;pTcpHandle->uAllLen = 0;}}}}}delete [] recv_buffer;recv_buffer = nullptr;return nullptr;
}TcpClient::TcpClient()
{pTcpHandle = new TcpHandle;epoll_fd = epoll_create(1);
}int32_t TcpClient::create_tcpClient(char *serverIp, int32_t serverPort)
{if (pTcpHandle == NULL)		return -1;pTcpHandle->fd = -1;if((pTcpHandle->fd = socket(AF_INET, SOCK_STREAM, 0)) < 0){printf("socket err=%s\n",strerror(errno));return -2;}pTcpHandle->remote_addr.sin_family = AF_INET;pTcpHandle->remote_addr.sin_port = htons(serverPort);pTcpHandle->remote_addr.sin_addr.s_addr = inet_addr(serverIp);if(connect(pTcpHandle->fd, (struct sockaddr *)&pTcpHandle->remote_addr, sizeof(pTcpHandle->remote_addr)) < 0){printf("connect err=%s\n",strerror(errno));return -3;}struct epoll_event evt;evt.events = EPOLLIN;fcntl(pTcpHandle->fd, F_SETFL, O_NONBLOCK);//设置非阻塞evt.data.fd = pTcpHandle->fd;epoll_ctl(epoll_fd,EPOLL_CTL_ADD,pTcpHandle->fd,&evt);m_runing = true;pthread_create(&threadId,NULL,DealTcpThread,this);return 0;
}int32_t TcpClient::SendData(char *data, int32_t len)
{int32_t ret = send(pTcpHandle->fd, data, len, MSG_NOSIGNAL);return ret;
}

TCP服务端

服务端启动监听，当有客户端接入时，向客户端循环发送大小不相等的数据包。
TcpServer.h

#ifndef TCPSERVER_H
#define TCPSERVER_H#include <string.h>
#include <stdint.h>
#include <stdint.h>
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <new>#define MAX_PKT_SIZE        (256<<20)   //网络包最大长度
//业务包头
struct CommMsgHdr
{uint16_t uMsgType;uint32_t uTotalLen;
};typedef struct _TcpHandle_{int32_t fd;uint32_t     uRcvLen;        //已接收数据大小uint32_t     uAllLen;        //消息总长度struct sockaddr_in local_addr;struct sockaddr_in remote_addr;_TcpHandle_(){uRcvLen = 0;uAllLen = 0;}
}TcpHandle;class TcpServer
{
public:TcpServer();int32_t create_tcpServer(int32_t listenPort);bool m_runing;int epoll_fd;TcpHandle* pTcpSerHandle;
private:pthread_t threadId;
};#endif // TCPSERVER_H

TcpServer.cpp

#include "TcpServer.h"int SendLoop(int32_t fd, uint8_t * buff, uint32_t len) {uint64_t total_send_bytes = 0;int64_t curr_send_len = 0;uint64_t left_bytes = len;while(total_send_bytes < len) {curr_send_len = send(fd, buff + total_send_bytes, left_bytes, MSG_NOSIGNAL);if(curr_send_len < 0) {if( errno == EINTR || errno == EAGAIN)continue;return -1;} else {total_send_bytes += curr_send_len;left_bytes -= curr_send_len;}}return 0;
}void* DealTcpThread(void* obj)
{TcpServer* pTcpServer = (TcpServer*)obj;TcpHandle* pTcpSerHandle = (TcpHandle*)pTcpServer->pTcpSerHandle;socklen_t src_len = sizeof(struct sockaddr_in);while (pTcpServer->m_runing){struct sockaddr_in src;memset(&src, 0, src_len);int connfd = accept(pTcpSerHandle->fd, (struct sockaddr*) &src, &src_len);if(connfd > -1){//开始发送for(int index=0;index<100;index++){uint32_t dataLength = 1024*1024*16 + index*10;void *sendbuff = new char[dataLength];CommMsgHdr* pHead = (CommMsgHdr*)sendbuff;pHead->uMsgType = 1001;pHead->uTotalLen = dataLength;SendLoop(connfd,(uint8_t * )sendbuff,dataLength);}}}return nullptr;
}TcpServer::TcpServer()
{pTcpSerHandle = new TcpHandle;
}int32_t TcpServer::create_tcpServer(int32_t listenPort)
{pTcpSerHandle->fd = -1;pTcpSerHandle->local_addr.sin_family = AF_INET;pTcpSerHandle->local_addr.sin_port = htons(listenPort);pTcpSerHandle->local_addr.sin_addr.s_addr = INADDR_ANY;pTcpSerHandle->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);int opt = 1;setsockopt(pTcpSerHandle->fd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));//复用端口if (bind(pTcpSerHandle->fd, (struct sockaddr*) &pTcpSerHandle->local_addr,sizeof(struct sockaddr_in)) < 0){printf("http server bind error(%s)",strerror(errno));return -1;}listen(pTcpSerHandle->fd, 32);m_runing = true;pthread_create(&threadId,NULL,DealTcpThread,this);return 0;
}

源码测试

先启动服务端

	TcpServer *pTcpServer;pTcpServer = new TcpServer;pTcpServer->create_tcpServer(9090);

再启动客户端

	TcpClient* pTcpClient;pTcpClient = new TcpClient;pTcpClient->create_tcpClient("127.0.0.1",9090);

客户端打印

Rcv completed,msgType=1001,uTotalLen=16777216
Rcv completed,msgType=1001,uTotalLen=16777226
Rcv completed,msgType=1001,uTotalLen=16777236
Rcv completed,msgType=1001,uTotalLen=16777246
Rcv completed,msgType=1001,uTotalLen=16777256
Rcv completed,msgType=1001,uTotalLen=16777266
Rcv completed,msgType=1001,uTotalLen=16777276
Rcv completed,msgType=1001,uTotalLen=16777286
...
...
...