C++实现udp分包和组包

udp开发中的几个问题

1、udp数据是怎么发送的

UDP（user datagram protocol，用户数据报协议）是无连接的，面向消息的。不会使用块的合并优化算法，由于UDP支持的是一对多的模式，所以接收端的skbuff(套接字缓冲区）采用了链式结构来记录每一个到达的UDP包，在每个UDP包中就有了消息头（消息来源地址，端口等信息）和结束标志， 即面向消息的通信是有消息保护边界的。
因此UDP是不会出现粘包的，但是会丢包。

2、tcp的处理方式

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

3、udp的MTU值

MTU = 1500 - IP头(20) - UDP头(8) = 1472(Bytes)（局域网）
MTU = 576 - 20 - 8 = 548（Internet互联网标准）
如果发送数据包大于MTU，则会出现严重丢包，甚至sendto返回失败。

4、udp大于MTU值怎么处理

1、UDP发送的数据报大小 <= MTU时，发送端直接发送，接收端无需组包。
2、UDP发送的数据报大小 > MTU时，发送端必须进行拆包发送，接收端收到包以后组包。

udp分包和组包策略

1、给每个整包分配一个唯一的序列号sequence，组包时根据序列号判断分包属于哪个整包。
2、每个子包分配一个index序号，用于接收端按顺序组包。
3、udp包是独立的，因此分包后，每个包都要有可识别的公共包头。
4、等所有分包都接收完成再进行组包。
5、udp是不可靠传输，如果分包的其中一个子包丢了，那么整个包将被丢弃（重传依赖其他响应机制）。
6、udp是无序的，发送端按序发，接收端收到包可能是乱序的，组包时要按顺序组包。
7、udp支持一对多通信，如果同时接收多个客户端的消息，多个客户端的消息会交叉到达，需要单独处理每个客户端的消息。

6 和 7 是实现的难点。

C++实现udp分包

写个udp客户端，往指定udp服务端发送数据，for循环连续发送多个包，发送间隔依据机器性能，性能好的机器可以无间隔发送（nosleep）还能保证不丢包。
可启动多个客户端同时向服务端发送数据，测试组包。

客户端实现：

#ifndef UDPCLIENT_H
#define UDPCLIENT_H

#include "SocketInclude.h"
class UdpClient
{
public:
    UdpClient();
    ~UdpClient();
    int CreateUdpCli(uint32_t serverIp, uint16_t _uListenPort);
    int dealUdpSendData();
private:
    UdpClientDef*    pUdpClientDef;
};
#endif //UDPCLIENT_H

#include "UdpClient.h"
#include <string>

UdpClient::UdpClient()
{
    pUdpClientDef = new UdpClientDef;
}

UdpClient::~UdpClient()
{
    delete pUdpClientDef;
}

//本端是客户端，udp客户端建链
int UdpClient::CreateUdpCli(uint32_t serverIp,uint16_t _uListenPort)
{
    return CreateUdpClient(pUdpClientDef,serverIp,_uListenPort);
}

//处理udp数据发送
int UdpClient::dealUdpSendData()
{
    static unsigned int sequence_whole = 0;//整包的序号
    //原始数据包
    uint32_t dataLength = 5000;
    void *sendbuff = new char[dataLength];
    CommMsgHdr* pHead = (CommMsgHdr*)sendbuff;

    pHead->uMsgType = 631;
    pHead->uTotalLen = dataLength;

    //开始分包
    int packetNum = dataLength / UDP_PAYLOAD;//分包数量
    int lastPaketSize = dataLength % UDP_PAYLOAD;//最后一片包的大小
    int sequence = 0;//当前发送的包序号
    if (lastPaketSize != 0)
    {
        packetNum = packetNum + 1;
    }

    //分包的包头
    MergeHdr tMergeHdr;
    tMergeHdr.uAllPktSize = dataLength;//负载总大小
    tMergeHdr.uPieces = packetNum;//分包数量
    tMergeHdr.uSequence = sequence_whole++;
    tMergeHdr.msgHead.uMsgType = 635;

    unsigned char piecebuff[UDP_PAYLOAD + sizeof(MergeHdr)];
    while (sequence < packetNum)
    {
        memset(piecebuff,0,UDP_PAYLOAD + sizeof(MergeHdr));
        if (sequence < (packetNum-1))//不是最后一片
        {
            tMergeHdr.uCurPktSize = sizeof(MergeHdr) + UDP_PAYLOAD;//当前包大小
            tMergeHdr.uIndex = sequence + 1;//当前包序号
            tMergeHdr.uOffset = sequence * UDP_PAYLOAD;//当前包偏移
            tMergeHdr.msgHead.uTotalLen  = tMergeHdr.uCurPktSize;
            memcpy(piecebuff, &tMergeHdr, sizeof(MergeHdr));
            memcpy(piecebuff+sizeof(MergeHdr), (char*)sendbuff+tMergeHdr.uOffset, UDP_PAYLOAD);

            ssize_t send_len = ::sendto(pUdpClientDef->fd, (const char*)piecebuff, tMergeHdr.uCurPktSize, 0, (struct sockaddr*) &pUdpClientDef->remote_addr,sizeof(struct sockaddr));
            if(send_len!=tMergeHdr.uCurPktSize)
            {
                printf("udp send failed,errno=[%d]\n",errno);
            }

            sequence ++;
        }
        else//最后一片
        {
            tMergeHdr.uCurPktSize = sizeof(MergeHdr)+(dataLength - sequence * UDP_PAYLOAD);
            tMergeHdr.uIndex = sequence + 1;
            tMergeHdr.uOffset = sequence * UDP_PAYLOAD;
            tMergeHdr.msgHead.uTotalLen  = tMergeHdr.uCurPktSize;
            memcpy(piecebuff, &tMergeHdr, sizeof(MergeHdr));
            memcpy(piecebuff+sizeof(MergeHdr), (char*)sendbuff+tMergeHdr.uOffset, dataLength - sequence*UDP_PAYLOAD);
            ssize_t send_len = ::sendto(pUdpClientDef->fd, (const char*)piecebuff, tMergeHdr.uCurPktSize, 0, (struct sockaddr*) &pUdpClientDef->remote_addr,sizeof(struct sockaddr));
            if(send_len!=tMergeHdr.uCurPktSize)
            {
                printf("udp send failed,errno=[%d]\n",errno);
            }

            sequence ++;
        }
    }
    delete[] (char *)(sendbuff), sendbuff = NULL;
    return 0;
}

结构体定义：

#ifndef SOCKETINCLUE_H
#define SOCKETINCLUE_H

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h> // for open
#include <unistd.h> // for close
#include <map>
#include <netinet/tcp.h>
#include <string.h>
#include <vector>

#define UDP_MTU             1472
#define MAX_EPOLL_EVENTS    1024         //epoll监听最大事件数量，对应最大连接socket数量

#pragma pack(push, 1)

//业务包头
struct CommMsgHdr
{
    uint16_t uMsgType;
    uint32_t uTotalLen;
};

//分包包头
struct MergeHdr
{
    CommMsgHdr  msgHead;//公共包头
    unsigned int uSequence;//整包的序号
    unsigned int uCurPktSize;//当前包的大小(sizeof(MergeHdr)+负载长度)
    unsigned int uAllPktSize;//数据的总大小
    unsigned int uPieces;//数据被分成包的个数
    unsigned int uIndex;//数据包当前的帧号
    unsigned int uOffset;//数据包在整个数据中的偏移
};
#define UDP_PAYLOAD         ( UDP_MTU - sizeof(MergeHdr) )    //互联网udp有效负载

//单个Sequence的所有包
struct pkt_merge
{
    void* mergebuff        = nullptr ;  //合并后的数据
    int  uAllPktSize       = 0;     //接收总长度
    unsigned int uSequence = 0;     //整包的序号
    unsigned int uPieces;           //数据被分成包的个数

    std::map<unsigned int,void*> mRcvData;//暂存接收的包<index,data>
};

//udp客户端
typedef struct _UdpClientDef_{
    int32_t             fd;              //fd，如果是服务端则统一使用服务端的fd，根据地址区分不同客户端
    struct sockaddr_in  remote_addr;     //对端地址
    std::map<int,pkt_merge*> m_pkt_merge;//<sequence,pkt_merge>,需要组包时，暂存包
}UdpClientDef;

//udp服务端
typedef struct _UdpServerDef_{
    int32_t fd;                         //fd
    struct sockaddr_in local_addr;      //本端地址
}UdpServerDef;

#pragma pack(pop)

int32_t CreateUdpClient(UdpClientDef* udp, uint32_t remoteIp, int32_t remotePort);
int32_t CreateUdpServer(UdpServerDef *udp, int32_t plocalPort);

#endif // SOCKETINCLUE_H

#include "SocketInclude.h"

//创建udp套接字
int32_t CreateUdpClient(UdpClientDef* udp, uint32_t remoteIp, int32_t remotePort)
{
    if (udp == NULL)		return -1;
    udp->fd = -1;

    udp->fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
    fcntl(udp->fd, F_SETFL, O_NONBLOCK);//设置非阻塞
    if(udp->fd < 0)
    {
        printf("[CreateUdpClient] create udp socket failed,errno=[%d],remoteIp=[%u],remotePort=[%d]",errno,remoteIp,remotePort);
        return -1;
    }

    udp->remote_addr.sin_family = AF_INET;
    udp->remote_addr.sin_port = htons(remotePort);
    udp->remote_addr.sin_addr.s_addr = remoteIp;
    return 0;
}

int32_t CreateUdpServer(UdpServerDef *udp, int32_t plocalPort)
{
    if (udp == NULL)		return -1;
    udp->fd = -1;

    udp->local_addr.sin_family = AF_INET;
    udp->local_addr.sin_port = htons(plocalPort);
    udp->local_addr.sin_addr.s_addr = INADDR_ANY;

    udp->fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
    if(udp->fd < 0)
    {
        printf("[CreateUdpServer] create udp socket failed,errno=[%d],plocalPort=[%d]",errno,plocalPort);
        return -1;
    }

    //2.socket参数设置
    int opt = 1;
    setsockopt(udp->fd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));//chw
    fcntl(udp->fd, F_SETFL, O_NONBLOCK);//设置非阻塞

    if (bind(udp->fd, (struct sockaddr*) &udp->local_addr,sizeof(struct sockaddr_in)) < 0)
    {
        close(udp->fd);
        printf("[CreateUdpServer] Udp server bind failed,errno=[%d],plocalPort=[%d]",errno,plocalPort);
        return -2;
    }

    return 0;
}

调用方式：

UdpClient mUdpClient;
uint32_t serverIp = inet_addr("127.0.0.1");
mUdpClient.CreateUdpCli(serverIp,9090);

for(int index=0;index<ui->lineEdit->text().toInt();index++)
{
        mUdpClient.dealUdpSendData();
        usleep(1000 * 10);
}

C++实现udp组包

写个udp服务端，绑定端口，使用epoll监听接收。文章来源地址https://www.toymoban.com/news/detail-597503.html

#ifndef UDPSERVER_H
#define UDPSERVER_H

#include "SocketInclude.h"
using namespace std;
class UdpServer
{
public:
    UdpServer();
    ~UdpServer();
    bool StartUp(uint16_t _uListenPort);//启动服务
private:
    static void* ThreadCallBack(void *arg);//创建线程函数
    void DealUdpThread();//udp线程处理函数
    void TryMergePkt(int index, unsigned int seq);//尝试组包
    void DelMergePkt(int index, unsigned int seq);//释放包
    void SetUdpEpollFlag(int fd, bool flag);

private:
    UdpServerDef*    pUdpServerDef;    //udp服务端对象
    pthread_t        m_threadId;    //udp线程ID
    int              m_epoll_fd;    //epollfd
    bool             m_bIsRunning;  //线程是否运行
    vector<UdpClientDef> m_vUdpClientDef;
};
#endif //UDPSERVER_H

#include "UdpServer.h"
#include <string>
#include <sys/epoll.h>
#include <QDebug>

UdpServer::UdpServer()
{
    pUdpServerDef = new UdpServerDef;
    m_epoll_fd = epoll_create(1);
    m_bIsRunning = true;
}

UdpServer::~UdpServer()
{
    m_bIsRunning = false;
}

bool UdpServer::StartUp(uint16_t _uListenPort)
{
    if(CreateUdpServer(pUdpServerDef,_uListenPort) == 0)
    {
        SetUdpEpollFlag(pUdpServerDef->fd,true);
    }

    if(pthread_create(&m_threadId, nullptr, ThreadCallBack, this))
    {
        printf("[UdpServer::StartUp] create thread failed.");
        return false;
    }

    return true;
}

//TCP服务监听线程，处理接入监听，客户端断开/错误管理
void* UdpServer::ThreadCallBack(void *arg)
{
    UdpServer *tm = static_cast<UdpServer *>(arg);
    if(tm)
        tm->DealUdpThread();
    return nullptr;
}

void UdpServer::DealUdpThread()
{
    const int kEpollDefaultWait = 1;//超时时长，单位ms
    struct epoll_event alive_events[MAX_EPOLL_EVENTS];

    void* bigBuffer = NULL;
    int  mergeRcvLen = 0;
    static unsigned int sequence = 1;
    while (m_bIsRunning)
    {
        int num = epoll_wait(m_epoll_fd, alive_events, MAX_EPOLL_EVENTS, kEpollDefaultWait);
        for (int i = 0; i < num; ++i)
        {
            int fd = alive_events[i].data.fd;
            int events = alive_events[i].events;

            if ( events & EPOLLIN )
            {
                char recv_buffer[UDP_MTU];
                memset(recv_buffer,0,UDP_MTU);
                ssize_t recv_len = 0;
                socklen_t src_len = sizeof(struct sockaddr_in);
                struct sockaddr_in SrcAddr;
                memset(&SrcAddr, 0, src_len);

                //1.开始接收
                struct sockaddr_in remote_addr;
                if ((recv_len = recvfrom(fd, recv_buffer, UDP_MTU, 0,	(struct sockaddr*) &SrcAddr, &src_len)) > 0)
                {
                    remote_addr.sin_port = SrcAddr.sin_port;
                    remote_addr.sin_addr.s_addr = SrcAddr.sin_addr.s_addr;

                    //判断是否已记录该客户端
                    bool isExist = false;
                    for(int index=0;index<m_vUdpClientDef.size();index++)
                    {
                        if(m_vUdpClientDef[index].remote_addr.sin_addr.s_addr == remote_addr.sin_addr.s_addr
                        && m_vUdpClientDef[index].remote_addr.sin_port        == remote_addr.sin_port)
                        {
                            isExist = true;
                            break;
                        }
                    }
                    if(!isExist)
                    {
                        UdpClientDef tUdpClientDef;
                        tUdpClientDef.fd = fd;
                        tUdpClientDef.remote_addr = remote_addr;
                        m_vUdpClientDef.push_back(tUdpClientDef);
                    }
                }
                else
                    continue;

                //1.不需要组包
                if( ((CommMsgHdr *)recv_buffer)->uMsgType != 635)
                {
                    //直接分发处理
                }

                //2.需要组包，处理错序
                for(int index=0;index<m_vUdpClientDef.size();index++)
                {
                    if(m_vUdpClientDef[index].remote_addr.sin_addr.s_addr == remote_addr.sin_addr.s_addr
                    && m_vUdpClientDef[index].remote_addr.sin_port        == remote_addr.sin_port)
                    {
                        MergeHdr* tMergeHdr = (MergeHdr*)recv_buffer;

                        if(m_vUdpClientDef[index].m_pkt_merge.count(tMergeHdr->uSequence) > 0)
                        {
                            //已存在该Sequence的包
                            m_vUdpClientDef[index].m_pkt_merge[tMergeHdr->uSequence]->uAllPktSize += tMergeHdr->uCurPktSize - sizeof(MergeHdr);;
                        }
                        else
                        {
                            //新Sequence的包
                            //建议新增定时器，在一定时间内没有收完包，则丢弃包
                            pkt_merge *tpkt_merge = new pkt_merge;
                            tpkt_merge->uSequence = tMergeHdr->uSequence;
                            tpkt_merge->uAllPktSize = tMergeHdr->uCurPktSize - sizeof(MergeHdr);
                            tpkt_merge->uPieces = tMergeHdr->uPieces;
                            tpkt_merge->mergebuff = new char[tMergeHdr->uAllPktSize];

                            m_vUdpClientDef[index].m_pkt_merge[tMergeHdr->uSequence] = tpkt_merge;
                        }
                        //todo,如果已存在key,存在泄漏
                        void* buff = new char[UDP_MTU];
                        memcpy(buff,recv_buffer,UDP_MTU);
                        m_vUdpClientDef[index].m_pkt_merge[tMergeHdr->uSequence]->mRcvData[tMergeHdr->uIndex] = buff;

                        TryMergePkt(index,tMergeHdr->uSequence);
                    }
                }

#if 0         //3.需要组包，不处理错序,此时如果同时接收两个客户端的数据，则可能出现错序
              if( ((CommMsgHdr *)recv_buffer)->uMsgType == 635)
                {
                    MergeHdr* tMergeHdr = (MergeHdr*)recv_buffer;
                    //根据sequence按顺序接收
                    if(sequence == tMergeHdr->uIndex)
                    {
                        sequence++;
                        if(bigBuffer == nullptr)
                            bigBuffer = new char[tMergeHdr->uAllPktSize];
                        mergeRcvLen += tMergeHdr->uCurPktSize - sizeof(MergeHdr);

                        memcpy((char*)bigBuffer+tMergeHdr->uOffset, recv_buffer + sizeof(MergeHdr),
                               tMergeHdr->uCurPktSize - sizeof(MergeHdr));
                        if ((tMergeHdr->uPieces == tMergeHdr->uIndex)
                                && (mergeRcvLen == tMergeHdr->uAllPktSize))
                        {
                            //组包完成
                            CommMsgHdr* pMsg = (CommMsgHdr*)bigBuffer;
                            printf("pMsg.uMsgType=%d\n",pMsg->uMsgType);
                            printf("pMsg.uTotalLen=%d\n",pMsg->uTotalLen);
                            printf("remote_addr.sin_port=%d\n",ntohs(remote_addr.sin_port));

                            mergeRcvLen = 0;
                            delete[] (char *)(bigBuffer);
                            bigBuffer = NULL;
                            sequence = 1;
                        }
                    }
                    //如果出现错序或乱序，丢弃包
                    else
                    {
                        mergeRcvLen = 0;
                        delete[] (char *)(bigBuffer);
                        bigBuffer = NULL;
                        sequence = 1;
                        printf(" miss-sequence \n");
                    }
                }
#endif
            }
        }

    }
}

//尝试组包
void UdpServer::TryMergePkt(int index, unsigned int seq)
{
    pkt_merge *tpkt_merge = m_vUdpClientDef[index].m_pkt_merge[seq];
    if(tpkt_merge->uPieces > tpkt_merge->mRcvData.size())
    {
        //还没收完所有的包，不组包
        auto last_ite = tpkt_merge->mRcvData.end();
        last_ite --;
        //如果此时收到的 分包序列号 >= 包的总数，则说明中间有丢包，此时丢弃包
        if(last_ite->first >= tpkt_merge->uPieces)
        {
            DelMergePkt(index, seq);
            printf("error,miss sequence\n");
        }
    }
    else if(tpkt_merge->uPieces < tpkt_merge->mRcvData.size())
    {
        //收到包的数量大于分片数量，出现异常，丢弃包
        printf("error,recv too many bags\n");
        DelMergePkt(index, seq);
    }
    else
    {
        //已经收完所有分包，开始组包
        auto ite = tpkt_merge->mRcvData.begin();
        while(ite != tpkt_merge->mRcvData.end())
        {
            MergeHdr* tMergeHdr = (MergeHdr*)ite->second;
            memcpy((char*)tpkt_merge->mergebuff+tMergeHdr->uOffset, (char*)ite->second + sizeof(MergeHdr),
                   tMergeHdr->uCurPktSize - sizeof(MergeHdr));
            ++ ite;
        }

        ite --;
        MergeHdr* tMergeHdr = (MergeHdr*)ite->second;
        if ((tpkt_merge->uPieces == tMergeHdr->uIndex)
                && (tpkt_merge->uAllPktSize == tMergeHdr->uAllPktSize))
        {
            //组包完成，分发处理
            CommMsgHdr* pMsg = (CommMsgHdr*)tpkt_merge->mergebuff;
            printf("pMsg.uMsgType=%d\n",pMsg->uMsgType);
            printf("pMsg.uTotalLen=%d\n",pMsg->uTotalLen);

            //释放资源
            DelMergePkt(index, seq);
        }
    }
}

void UdpServer::DelMergePkt(int index, unsigned int seq)
{
    pkt_merge *tpkt_merge = m_vUdpClientDef[index].m_pkt_merge[seq];

    auto ite_2 = tpkt_merge->mRcvData.begin();
    while(ite_2 != tpkt_merge->mRcvData.end())
    {
        delete[] (char *)(ite_2->second);
        ite_2++;
    }
    delete[] (char *)(tpkt_merge->mergebuff);
    tpkt_merge->mRcvData.clear();
    m_vUdpClientDef[index].m_pkt_merge.erase(m_vUdpClientDef[index].m_pkt_merge.find(tpkt_merge->uSequence));
}

//设置epoll监听udp套接字，只监听EPOLLIN事件
void UdpServer::SetUdpEpollFlag(int fd, bool flag)
{
    struct epoll_event evt;
    evt.events = EPOLLIN;
    evt.data.fd = fd;
    if(flag)
        epoll_ctl(m_epoll_fd,EPOLL_CTL_ADD,fd,&evt);
    else
        epoll_ctl(m_epoll_fd,EPOLL_CTL_DEL,fd,&evt);
}

到了这里，关于C++实现udp分包和组包的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！