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;//超时时长，单位ms
struct 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);//关闭socket
continue;
}
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);//关闭socket
continue;
}
//如果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);//关闭socket
continue;
}
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
...
...
...
