什么是惊群
简单说,惊群是因为多进程(多线程)在同时阻塞等待同一个事件的时候(休眠状态),当时间发生时,就会唤醒所有等待的(休眠的)进程(线程)。但是事件只能被一个进程或线程处理,而其他进程(线程)获取失败,只能重新进入休眠状态,这种现象和性能浪费就叫做惊群。维基百科-惊群
产生惊群的条件
- 多个进程或者多个线程
- 同时等待处理一个事件
具体场景复现
测试环境 :
- 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);
tfd = accept(socketFd, (sockaddr *) &cli_addr, &length);
std::cout << "worker" << k << " in " << std::endl;
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个线程都白白唤醒浪费了性能
为啥内核已经处理了第一种情景下的惊群问题,第二种情景下的惊群问题为啥就不处理了呢?
我的猜想不一定正确,如果有错误,请指出:
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,限制了真正并行的吞吐量。 这个有待验证,等我有时间再去深入了解一下。
今天的文章linux 多线程或多进程 epoll处理 accept 惊群问题分享到此就结束了,感谢您的阅读。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/22943.html