快捷導(dǎo)航

C++解決TCP粘包的問題實(shí)現(xiàn)

更新時間：2023年08月07日 09:37:05 作者：夏天匆匆2過

本文主要介紹了C++解決TCP粘包的問題實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

TCP粘包問題

TCP是面向連接的，面向流的可靠性傳輸。TCP會將多個間隔較小且數(shù)據(jù)量小的數(shù)據(jù)，合并成一個大的數(shù)據(jù)塊，然后進(jìn)行封包發(fā)送，這樣一個數(shù)據(jù)包里就可能含有多個消息的數(shù)據(jù)，面向流的通信是無消息保護(hù)邊界的，也就是TCP粘包。接收端需要自己完成數(shù)據(jù)的拆包和組包，解決粘包問題。

要解決TCP粘包問題，就要給TCP定義公共包頭，包頭一般包括消息類型和消息大小，用包頭來分割每個數(shù)據(jù)包，做數(shù)據(jù)包的邊界。

下面分別用C++實(shí)現(xiàn)TCP客戶端和TCP服務(wù)端，使用qt測試。

TCP客戶端

TCP客戶端主動連接到TCP服務(wù)端，并接收TCP服務(wù)端發(fā)送的數(shù)據(jù)，對接收的數(shù)據(jù)按照定義的公共包頭進(jìn)行分割組包，每當(dāng)組成一個完整數(shù)據(jù)包時，打印相關(guān)信息。

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) ? //網(wǎng)絡(luò)包最大長度
//業(yè)務(wù)包頭
struct CommMsgHdr
{
? ? uint16_t uMsgType;
? ? uint32_t uTotalLen;
};
typedef struct _TcpHandle_{
? ? int32_t fd;
? ? uint32_t ? ? uRcvLen; ? ? ? ?//已接收數(shù)據(jù)大小
? ? 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);//關(guān)閉socket
? ? ? ? ? ? ? ? ? ? ? ? continue;
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? pTcpHandle->uRcvLen += iRecvLen;
? ? ? ? ? ? ? ? ? ? //如果已經(jīng)接收完整頭部
? ? ? ? ? ? ? ? ? ? if(pTcpHandle->uRcvLen >= head_len)
? ? ? ? ? ? ? ? ? ? {
? ? ? ? ? ? ? ? ? ? ? ? CommMsgHdr* pHdr = (CommMsgHdr *)recv_buffer;
? ? ? ? ? ? ? ? ? ? ? ? pTcpHandle->uAllLen = pHdr->uTotalLen;
? ? ? ? ? ? ? ? ? ? ? ? //如果報(bào)文頭里的uTotalLen太小或太大，異常處理
? ? ? ? ? ? ? ? ? ? ? ? if ( pHdr->uTotalLen < head_len || pHdr->uTotalLen > MAX_PKT_SIZE )
? ? ? ? ? ? ? ? ? ? ? ? {
? ? ? ? ? ? ? ? ? ? ? ? ? ? printf("uTotalLen invalid,uTotalLen=%u,fd=[%d]",
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?pHdr->uTotalLen,fd);
? ? ? ? ? ? ? ? ? ? ? ? ? ? close(fd);//關(guān)閉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.開始接收數(shù)據(jù)體
? ? ? ? ? ? ? ? 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);//關(guān)閉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);//設(shè)置非阻塞
? ? 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服務(wù)端

服務(wù)端啟動監(jiān)聽，當(dāng)有客戶端接入時，向客戶端循環(huán)發(fā)送大小不相等的數(shù)據(jù)包。

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) ? //網(wǎng)絡(luò)包最大長度
//業(yè)務(wù)包頭
struct CommMsgHdr
{
? ? uint16_t uMsgType;
? ? uint32_t uTotalLen;
};
typedef struct _TcpHandle_{
? ? int32_t fd;
? ? uint32_t ? ? uRcvLen; ? ? ? ?//已接收數(shù)據(jù)大小
? ? 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)
? ? ? ? {
? ? ? ? ? ? //開始發(fā)送
? ? ? ? ? ? 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));//復(fù)用端口
? ? 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;
}

源碼測試

先啟動服務(wù)端

    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
...
...
...

到此這篇關(guān)于C++解決TCP粘包的問題實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)C++ TCP粘包內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: