[系统运维] linux 多线程或多进程 epoll处理 accept 惊群问题

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 系统运维 -> linux 多线程或多进程 epoll处理 accept 惊群问题 -> 正文阅读

[系统运维]linux 多线程或多进程 epoll处理 accept 惊群问题

什么是惊群

简单说，惊群是因为多进程（多线程）在同时阻塞等待同一个事件的时候（休眠状态），当时间发生时，就会唤醒所有等待的（休眠的）进程（线程）。但是事件只能被一个进程或线程处理，而其他进程（线程）获取失败，只能重新进入休眠状态，这种现象和性能浪费就叫做惊群。

产生惊群的条件

多个进程或者多个线程
同时等待处理一个事件

具体场景复现

测试环境 :

debian11
内核5.10.0-8
编译器 clang11

多线程和多进程在惊群问题上差不多，为了少些一点，下文中没有特殊说明，多线程也包含了多进程

在linux中，使用C/C++ 编写 tcp server时，会依次调用 socket() bind() listen() accept() 这几个函数，这几个函数会打开socket，绑定ip和端口，开始监听端口，accept函数会阻塞当前进程，等待客户端连接。

如果在单线程中，只有一个accept函数在等待客户端连接，当客户端来连接的时候，只会有一个accept函数来处理，所以也不会存在惊群问题了。

在多线模型中，多个线程分别accept同一个socket，当有客户端连接时，内核会通知所有的线程来处理这个请求，但是呢，请求只能被一个线程处理，其他的线程的不到这个事件，只能白白被唤醒。

这是最简单的一种惊群，这种情况在linux2.6以后就不会产生了。因为在Linux 2.6 版本之后，通过引入一个标记位 WQ_FLAG_EXCLUSIVE，解决掉了 Accept 惊群效应。我原本还想在centos3.9（内核版本是2.5）中去复现这种情况，但是折腾了好久，也没能在centos上编译也运行C++程序，遂放弃。不废话了，上代码，测试第一种情况

#include <netinet/in.h>
#include <iostream>
#include <sys/epoll.h>
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>

#define WORKER_THREAD 4
//创建socket，并返回fd
int createSocket() {
    int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (fd < 0) {
        std::cout << "create socket error" << std::endl;
        return 0;
    }

    sockaddr_in sockAddr{};
    sockAddr.sin_port = htons(PORT);
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = htons(INADDR_ANY);

    if (bind(fd, (sockaddr *) &sockAddr, sizeof(sockAddr)) < 0) {
        std::cout << "bind socket error, port:" << PORT << std::endl;
        return 0;
    }

    if (listen(fd, 100) < 0) {
        std::cout << "listen port error" << std::endl;
        return 0;
    }
    return fd;
}

void Worker1(int socketFd, int k) {
    std::cout << " Worker" << k << " run " << std::endl;
    while (true) {
        int tfd = 0;
        sockaddr_in cli_addr{};
        socklen_t length = sizeof(cli_addr);
        std::cout << "worker" << k << " in " << std::endl;
        tfd = accept(socketFd, (sockaddr *) &cli_addr, &length);
        if (tfd <= 0) {
            std::cout << "accept error" << std::endl;
            return;
        } else {
            std::cout << "worker" << k << " accept " << std::endl;
        }
    }
}

int main() {
    std::mutex mutex;
    std::unique_lock<std::mutex> lck(mutex);
    std::condition_variable cv;
    
    int fd = createSocket();
    //第一种,多个线程不使用多路复用,accept同一个socket
    for (int i = 0; i < WORKER_THREAD; ++i) {
        std::thread th(&Worker1, fd, i + 1);
        th.detach();
    }

    cv.wait(lck);
    return 0;
}

这代码可以用C写，但是习惯用C++了，就用C++写吧。代码也比较简单，createSocket()创建了一个socket，然后4个线程分别去accept这个socket。下面是运行结果：

可以看到，4个线程都在运行，并且accept，但是当连接来的时候，只有个线程能得到事件。

既然linux内核已经帮我们处理了惊群，那我们还考虑这些干啥，直接用不就完了。

但是，我们在写代码的时候一般不会直接阻塞accept的，都是使用多路复用来帮我们处理连接阻塞的是多路复用函数。目前综合性能比较好的IO多路复用是epoll。当在多线程中使用epoll时，惊群问题就会出现了。先代码和结果，然后再解释

#include <netinet/in.h>
#include <iostream>
#include <sys/epoll.h>
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>

#define WORKER_THREAD 4
//创建socket，并返回fd
int createSocket() {
    int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (fd < 0) {
        std::cout << "create socket error" << std::endl;
        return 0;
    }

    sockaddr_in sockAddr{};
    sockAddr.sin_port = htons(PORT);
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = htons(INADDR_ANY);

    if (bind(fd, (sockaddr *) &sockAddr, sizeof(sockAddr)) < 0) {
        std::cout << "bind socket error, port:" << PORT << std::endl;
        return 0;
    }

    if (listen(fd, 100) < 0) {
        std::cout << "listen port error" << std::endl;
        return 0;
    }
    return fd;
}

void Worker2(int socketFd, int k) {
    std::cout << " Worker" << k << " run " << std::endl;

    int eFd = epoll_create(1);
    if (eFd < 0) {
        std::cout << "create epoll fail";
        return;
    }
    epoll_event epev_{};
    epev_.events = EPOLLIN;
    epev_.data.fd = socketFd;
    epoll_ctl(eFd, EPOLL_CTL_ADD, socketFd, &epev_);
    epoll_event events[EVENT_NUM];

    while (true) {
        int eNum = epoll_wait(eFd, events, EVENT_NUM, -1);
        if (eNum == -1) {
            std::cout << "epoll error";
            return;
        }
        //一定要加上这句,防止事件被瞬间处理,导致看不到结果
        std::this_thread::sleep_for((std::chrono::seconds (1)));
        std::cout << "worker" << k << " in " << std::endl;
        for (int i = 0; i < eNum; ++i) {
            if (events[i].data.fd == socketFd) {
                int tfd = 0;
                sockaddr_in cli_addr{};
                socklen_t length = sizeof(cli_addr);
                tfd = accept(socketFd, (sockaddr *) &cli_addr, &length);
                if (tfd <= 0) {
                    std::cout << "accept error" << std::endl;
                } else {
                    std::cout << "worker" << k << " accept " << std::endl;
                }
            } else {
                //处理正常的socket读写事件,这里可以忽略,不是这次关注的点
            }
        }
    }
}

int main() {
    std::mutex mutex;
    std::unique_lock<std::mutex> lck(mutex);
    std::condition_variable cv;
    int fd = createSocket();
    //第二种,多个线程使用epoll多路复用,accept同一个socket
    for (int i = 0; i < WORKER_THREAD; ++i) {
        std::thread th(&Worker2, fd, i + 1);
        th.detach();
    }
}

结果

这里可以看到，当有客户端来连接的时候，4个线程都被唤醒了，但是只有workr2 线程成功获取了事件，其余的3个线程都白白唤醒浪费了性能

情景下的惊群问题，第二种情景下的惊群问题为啥就不处理了呢？

相关视频推荐

后台开发第400讲|6种epoll的做法，从redis，memcached到nginx的网络模型实现|1. epoll的封装模式 2. reactor的作用与原

需要C/C++ Linux服务器架构师学习资料加956314242群获取（资料包括C/C++，Linux，golang技术，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg等），免费分享

我的猜想不一定正确，如果有错误，请指出：

accept 只能是被一个进程调用成功（连接事件只会处理一次嘛），所以内核就直接处理了（一个accept只会唤醒一个进程）。但 epoll 不一样，epoll中管理了很多连接，不止socket这一个，除了可能后续被 accept 调用外，还有可能是其他网络 IO 事件的，而其他 IO 事件是否只能由一个进程处理，是不一定的，这是一个由用户决定的事情，例如可能一个文件会由多个进程来读写。所以，对 epoll 默认对于多进程监听同一文件不会设置互斥，所以就导致了epoll惊群问题。

在linux4.5内核之后给epoll添加了一个 EPOLLEXCLUSIVE的标志位，如果设置了这个标志位，那epoll将进程挂到等待队列时将会设置一下互斥标志位，这时实现跟内核原生accept一样的特性，只会唤醒队列中的一个进程。

修改一下worker2函数:

void Worker2(int socketFd, int k) {
    std::cout << " Worker" << k << " run " << std::endl;

    int eFd = epoll_create(1);
    if (eFd < 0) {
        std::cout << "create epoll fail";
        return;
    }
    epoll_event epev_{};
    //给epoll加上 互斥标志
    epev_.events = EPOLLIN | EPOLLEXCLUSIVE;
    epev_.data.fd = socketFd;
    epoll_ctl(eFd, EPOLL_CTL_ADD, socketFd, &epev_);
    epoll_event events[EVENT_NUM];

    while (true) {
        int eNum = epoll_wait(eFd, events, EVENT_NUM, -1);
        if (eNum == -1) {
            std::cout << "epoll error";
            return;
        }
        //一定要加上这句,防止事件被瞬间处理,导致看不到结果
        std::this_thread::sleep_for((std::chrono::seconds(1)));
        std::cout << "worker" << k << " in " << std::endl;
        for (int i = 0; i < eNum; ++i) {
            if (events[i].data.fd == socketFd) {
                int tfd = 0;
                sockaddr_in cli_addr{};
                socklen_t length = sizeof(cli_addr);
                tfd = accept(socketFd, (sockaddr *) &cli_addr, &length);
                if (tfd <= 0) {
                    std::cout << "accept error" << std::endl;
                } else {
                    std::cout << "worker" << k << " accept " << std::endl;
                }
            } else {
                //处理正常的socket读写事件,这里可以忽略,不是这次关注的点
            }
        }
    }
}

现在来测试一下

现在的epoll已经不会有惊群问题了

另一种方式

其实解决多线程使用epoll等多路复用导致的惊群问题，还有一个更彻底解决方法，让每个线程分别打开一个socket，并且这些socket绑定在同一个端口，然后accept这个socket。这就像第一种情景那样，内核直接帮我们做了惊群处理。这里会使用到 linux 3.9后 socket提供SO_REUSEPORT标志。使用这个标志后，会允许多个socket绑定和监听同一个端口。代码如下

#include <netinet/in.h>
#include <iostream>
#include <sys/epoll.h>
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>

#define WORKER_THREAD 4
//创建socket，并返回fd
int createSocket2() {
    int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (fd == -1) {
        std::cout << "create socket error" << std::endl;
        return 0;
    }

    int on = 1;
    if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, (const void *) &on, sizeof(on)) < 0) {
        std::cout << "set opt error, ret:" << std::endl;
    }

    sockaddr_in sockAddr{};
    sockAddr.sin_port = htons(PORT);
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = htons(INADDR_ANY);

    if (bind(fd, (sockaddr *) &sockAddr, sizeof(sockAddr)) < 0) {
        std::cout << "bind socket error, port:" << PORT << std::endl;
        return 0;
    }

    if (listen(fd, 100) < 0) {
        std::cout << "listen port error" << std::endl;
        return 0;
    }
    return fd;
}

void Worker3(int k) {
    std::cout << " Worker" << k << " run " << std::endl;

    int socketFd = createSocket2();
    int eFd = epoll_create(1);
    if (eFd == -1) {
        std::cout << "create epoll fail" << std::endl;
        return;
    }

    epoll_event epev_{};
    epev_.events = EPOLLIN;
    epev_.data.fd = socketFd;
    epoll_ctl(eFd, EPOLL_CTL_ADD, socketFd, &epev_);
    epoll_event events[EVENT_NUM];

    while (true) {
        int eNum = epoll_wait(eFd, events, EVENT_NUM, -1);
        if (eNum == -1) {
            std::cout << "epoll error" << std::endl;
            return;
        }
        std::this_thread::sleep_for((std::chrono::seconds(1)));
        std::cout << "worker" << k << " in " << std::endl;
        for (int i = 0; i < eNum; ++i) {
            if (events[i].data.fd == socketFd) {
                int tfd = 0;
                sockaddr_in cli_addr{};
                socklen_t length = sizeof(cli_addr);
                tfd = accept(socketFd, (sockaddr *) &cli_addr, &length);
                if (tfd <= 0) {
                    std::cout << "accept error" << std::endl;
                } else {
                    std::cout << "worker" << k << " accept " << std::endl;
                }
            } else {
                //处理正常的socket读写事件
            }
        }
    }
}

int main() {
    std::mutex mutex;
    std::unique_lock<std::mutex> lck(mutex);
    std::condition_variable cv;

    //第三种,多个线程使用epoll多路复用,每个线程分别bind,listen 同一个端口, accept各自的socket
    for (int i = 0; i < WORKER_THREAD; ++i) {
        std::thread th(&Worker3, i + 1);
        th.detach();
    }

    cv.wait(lck);
    return 0;
}

结果

也是没有问题的，多个连接来的时候，只会有一个线程被唤醒，相当于在内核级别中实现了一个负载均衡

总结

简单总结一下，当多个线程或者进程同时阻塞同一个事件的时候，会出现惊群现象，如果不适用epoll等多路复用技术，在linux2.6 以后内核已经帮我们处理了惊群问题。

如果使用了epoll，就需要额外处理epoll导致的惊群问题，有两种方式

linux4.5内核之后，epoll有一个EPOLLEXCLUSIVE特性，可以防止epoll惊群出现
linux 3.9内核之后给 socket 提供SO_REUSEPORT特性，可以允许多个socket绑定在同一个端口上，相当于每个线程都有一个socket，在处理accept时，内核会自动处理惊群问题

1和2两种方式都能有效解决惊群问题，但是目前使用 socket的 SO_REUSEPORT 是最好的方式.。

我通过查资料得到 EPOLLEXCLUSIVE 标识会保证一个事件发生时候只有一个线程会被唤醒，来避免多惊群问题。不过任一时候只能有一个Worker调用 accept，限制了真正并行的吞吐量。这个有待验证，小伙伴有时间可以去深入了解一下。