网站建设中颜色的感染力,wordpress评论开关,个人网站命名技巧,wordpress 主题和插件下载失败文章目录 1. Reactor模型2. Proactor模型3. 同步IO模拟Proactor模型 在高并发编程和网络连接的消息处理中#xff0c;通常可分为两个阶段#xff1a;等待消息就绪和消息处理。当使用默认的阻塞套接字时#xff08;例如每个线程专门处理一个连接#xff09;#xff0c;这两… 文章目录 1. Reactor模型2. Proactor模型3. 同步IO模拟Proactor模型 在高并发编程和网络连接的消息处理中通常可分为两个阶段等待消息就绪和消息处理。当使用默认的阻塞套接字时例如每个线程专门处理一个连接这两个阶段往往是合并的。因此处理套接字的线程需要等待消息就绪这在高并发场景下导致线程频繁地休眠和唤醒从而影响 CPU 使用效率。 为提高高并发编程效率可将等待消息就绪和消息处理两个阶段分开。换言之等待消息就绪的代码段与处理消息的代码段应分离。这要求套接字必须为非阻塞否则处理消息的代码段可能导致线程在条件不满足时进入等待状态。实现等待消息就绪阶段的方法是线程主动查询或让单个线程等待所有连接。这就引入了 I/O 多路复用技术它可以同时处理多个连接尽管仍可能需要等待并导致线程休眠但由于它可以监控所有连接所以当线程被唤醒时必定有一些连接已准备好进行处理。
高性能服务器程序通常需要处理三类事件I/O 事件定时事件以及信号。为实现高效的事件处理有两种主要模型Reactor 和 Proactor。这两种模型分别采用不同的策略来解决等待消息就绪和消息处理阶段的问题以提高服务器在高并发环境下的性能表现。
1. Reactor模型
首先让我们回顾一下常规函数调用的机制程序调用某个函数函数执行程序等待然后函数将结果和控制权返回给程序程序继续处理。与此不同Reactor反应堆是一种事件驱动的机制。它颠覆了传统的事件处理流程应用程序不再主动调用某个 API 完成处理而是将处理逻辑逆置。在这种模式下应用程序需要提供相应的接口并将其注册到 Reactor 上。当相应的事件发生时Reactor 会主动调用应用程序注册的接口这些接口通常被称为“回调函数”。通过这种方式Reactor 实现了事件驱动编程提高了程序的响应速度和效率。 Reactor 模式是处理并发 I/O 的一种常见模式适用于同步 I/O。其核心思想是将所有待处理的 I/O 事件注册到一个中心 I/O 多路复用器上同时让主线程/进程在多路复用器上阻塞。当有 I/O 事件到来或准备就绪如文件描述符或 socket 可读、写时多路复用器返回并将事先注册的相应 I/O 事件分发到对应的处理器中。这样Reactor 模式能够实现高效的并发 I/O 处理提高程序在高并发环境下的性能表现。
Reactor模型有三个重要的组件
ReactorReactor 是核心组件负责管理事件循环和I/O事件分发。它接收客户端连接监听I/O事件并将这些事件分发给相应的事件处理器。Reactor可以是单线程或多线程的根据具体应用场景和性能要求进行选择。Handlers事件处理器Handlers是用于处理特定I/O事件的对象。每个处理器通常对应一个客户端连接或一个资源如文件、套接字等。处理器负责处理与它们关联的事件如读取数据、处理业务逻辑、发送响应等。处理器可以是同步或异步的取决于具体的实现方式。Demultiplexer事件分离器Demultiplexer负责从操作系统中获取I/O事件并将这些事件传递给Reactor。常见的事件分离器实现方式有 select, poll, epoll, kqueue 等。事件分离器使Reactor能够同时处理多个I/O事件从而实现高并发性能。 事件处理的具体流程
Reactor 接收客户端连接将连接的相关I/O事件注册到事件分离器Demultiplexer上。同时Reactor 将事件处理器Handlers与这些I/O事件关联起来。事件分离器Demultiplexer负责检测并收集操作系统中发生的I/O事件。当有I/O事件发生时事件分离器将这些事件通知Reactor。Reactor 将收到的I/O事件与之前注册的事件处理器Handlers进行匹配。找到匹配的处理器后Reactor 将事件分发给相应的处理器。事件处理器Handlers收到分发的I/O事件后执行相应的操作如读取数据、处理业务逻辑、发送响应等。事件处理完成后Reactor 可以根据需要更新处理器与I/O事件之间的关联以便继续处理后续事件。
Reactor模型的优点
高并发性能Reactor模式通过事件驱动的方式实现了高并发性能。事件分离器可以同时检测多个I/O事件Reactor可以根据事件分离器的通知并行处理多个I/O事件从而提高服务器的并发处理能力。良好的可扩展性Reactor模式可以通过增加线程数量或使用多个Reactor实例来提高系统的可扩展性。在多核处理器环境下可以利用多线程Reactor模式有效地分摊负载进一步提高系统性能。资源利用率高Reactor模式中事件处理器只在有I/O事件发生时才执行操作。这样可以避免无效的轮询和资源浪费提高系统资源的利用率。易于管理和维护Reactor模式将事件处理器和I/O事件分离使得事件处理逻辑更加清晰和易于管理。此外Reactor模式还可以实现处理器的动态注册和注销便于系统的维护和扩展。可适应不同场景Reactor模式可以根据不同的应用场景和性能要求进行定制。例如可以选择单线程或多线程Reactor同步或异步处理器等以满足特定应用的需求。
Reactor 模型在开发效率方面相较于直接使用 IO 复用有所提高。该模型通常采用单线程设计目标是让单线程充分利用一颗 CPU 的全部资源。同时它还带有一个优点即在处理事件时很多情况下无需考虑共享资源的互斥访问问题。然而这个模型也存在明显的缺点。随着硬件的发展摩尔定律不再适用CPU 频率由于材料限制无法继续大幅提升因此提升性能主要依赖于增加核数。在需要利用多核资源的程序中Reactor 模型的表现可能会受到影响。
对于业务较简单的程序如仅需访问提供并发访问的服务可以直接启用多个反应堆每个反应堆对应一颗 CPU 核心。在这种情况下运行在各个反应堆上的请求互不相关从而充分利用多核资源。例如Nginx 这类的 HTTP 静态服务器就采用了这种策略。 在C里使用Reactor模型实现的TCP回显服务器
需要系统支持epoll并且编译器支持C11
#include iostream
#include cstdlib
#include cstring
#include cstdio
#include vector
#include map
#include unistd.h
#include sys/epoll.h
#include arpa/inet.h
#include fcntl.h
#include netinet/in.h
#include sys/socket.h#define MAX_EVENTS 10class Reactor {
public:Reactor() {_epoll_fd epoll_create1(0);if (_epoll_fd -1) {perror(epoll_create1);exit(EXIT_FAILURE);}}~Reactor() {close(_epoll_fd);}void add_fd(int fd, uint32_t events) {struct epoll_event event;event.data.fd fd;event.events events;if (epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, fd, event) -1) {perror(epoll_ctl);exit(EXIT_FAILURE);}}void del_fd(int fd) {if (epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, nullptr) -1) {perror(epoll_ctl);exit(EXIT_FAILURE);}}void run() {std::vectorstruct epoll_event events(MAX_EVENTS);while (true) {int n epoll_wait(_epoll_fd, events.data(), MAX_EVENTS, -1);if (n -1) {perror(epoll_wait);exit(EXIT_FAILURE);}for (int i 0; i n; i) {if (events[i].events EPOLLIN) {handle_input(events[i].data.fd);} else if (events[i].events EPOLLOUT) {handle_output(events[i].data.fd);}}}}virtual void handle_input(int fd) 0;virtual void handle_output(int fd) 0;private:int _epoll_fd;
};class EchoReactor : public Reactor {
public:EchoReactor(int listen_fd) : _listen_fd(listen_fd) {add_fd(_listen_fd, EPOLLIN);}void handle_input(int fd) override {if (fd _listen_fd) {struct sockaddr_in addr;socklen_t addrlen sizeof(addr);int conn_fd accept(_listen_fd, (struct sockaddr *)addr, addrlen);if (conn_fd -1) {perror(accept);exit(EXIT_FAILURE);}make_socket_non_blocking(conn_fd);add_fd(conn_fd, EPOLLIN);} else {char buf[1024];ssize_t n read(fd, buf, sizeof(buf));if (n 0) {if (n 0) perror(read);close(fd);del_fd(fd);} else {_out_buffers[fd] std::string(buf, n);del_fd(fd);add_fd(fd, EPOLLOUT);}}
}void handle_output(int fd) override {auto it _out_buffers.find(fd);if (it ! _out_buffers.end()) {ssize_t n write(fd, it-second.c_str(), it-second.size());if (n 0) {if (n 0) perror(write);close(fd);del_fd(fd);} else {it-second.erase(0, n);if (it-second.empty()) {del_fd(fd);add_fd(fd, EPOLLIN);}}}
}private:int make_socket_non_blocking(int sfd) {int flags fcntl(sfd, F_GETFL, 0);if (flags -1) {perror(fcntl);return -1;}flags | O_NONBLOCK;if (fcntl(sfd, F_SETFL, flags) -1) {perror(fcntl);return -1;}return 0;}int _listen_fd;std::mapint, std::string _out_buffers;
};int main(int argc, char *argv[]) {if (argc ! 2) {std::cerr Usage: argv[0] port std::endl;exit(EXIT_FAILURE);}int port std::stoi(argv[1]);int listen_fd socket(AF_INET, SOCK_STREAM, 0);if (listen_fd -1) {perror(socket);exit(EXIT_FAILURE);}int optval 1;if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, optval, sizeof(optval)) -1) {perror(setsockopt);exit(EXIT_FAILURE);}struct sockaddr_in addr;addr.sin_family AF_INET;addr.sin_port htons(port);addr.sin_addr.s_addr INADDR_ANY;if (bind(listen_fd, (struct sockaddr *)addr, sizeof(addr)) -1) {perror(bind);exit(EXIT_FAILURE);}if (listen(listen_fd, SOMAXCONN) -1) {perror(listen);exit(EXIT_FAILURE);}EchoReactor echo_reactor(listen_fd);echo_reactor.run();return 0;
}使用telnet测试
2. Proactor模型
在众多网络编程模型中Reactor模型因其高并发性能、资源利用率高等优点而受到广泛关注。Reactor模型基于事件驱动的原理通过Reactor、事件处理器Handlers和事件分离器Demultiplexer三个核心组件的协同作用有效地处理大量并发连接和I/O操作。然而在某些高负载场景下Reactor模型可能会遇到事件处理瓶颈的问题这使得我们需要寻求其他解决方案来进一步提升系统性能。
Proactor模型正是在这种背景下应运而生的一种高性能网络编程模型。与Reactor模型类似Proactor模型同样采用了事件驱动的方式但在事件处理的实现上有所不同。Proactor模型将异步I/O操作与事件处理器相结合使得I/O操作可以在后台执行从而进一步减少阻塞和提高系统性能。在Proactor模型中事件处理的主要责任由操作系统承担而应用程序则专注于处理业务逻辑。这种分工使得Proactor模型能够更好地应对高负载场景为构建高性能网络服务器提供了一种有力的解决方案。 Proactor模型有三个重要的组件
ProactorProactor是核心组件负责管理事件循环和I/O事件分发。它接收客户端连接监听I/O事件并将这些事件分发给相应的事件处理器。与Reactor模型相比Proactor将大部分I/O操作的处理交给操作系统从而进一步减少阻塞和提高系统性能。Asynchronous Handlers异步事件处理器Asynchronous Handlers是用于处理特定I/O事件的对象。每个处理器通常对应一个客户端连接或一个资源如文件、套接字等。与Reactor模型中的事件处理器不同Proactor模型中的处理器通过异步I/O操作与事件处理相结合使得I/O操作可以在后台执行进一步提高系统性能。Asynchronous Operation Processor异步操作处理器Asynchronous Operation Processor是负责执行实际的异步I/O操作的组件。它与事件处理器协同工作将I/O操作的执行与事件处理解耦进一步减少阻塞和提高系统性能。Completion Dispatcher完成分发器Completion Dispatcher负责在异步I/O操作完成后通知Proactor。当操作系统完成一个异步I/O操作时它会将完成通知发送给Completion Dispatcher而后者则将通知传递给Proactor。这样Proactor可以将事件分发给相应的事件处理器处理业务逻辑。 事件处理的具体流程
Proactor 接收客户端连接并将连接的相关I/O事件注册到异步事件处理器Asynchronous Handlers上。异步事件处理器Asynchronous Handlers启动异步I/O操作。此时I/O操作的执行由异步操作处理器Asynchronous Operation Processor负责与事件处理器解耦使得I/O操作可以在后台执行。异步操作处理器Asynchronous Operation Processor与操作系统协同工作执行实际的异步I/O操作。这样事件处理器可以在等待I/O操作完成的过程中处理其他任务进一步提高系统性能。当操作系统完成一个异步I/O操作时它会将完成通知发送给完成分发器Completion Dispatcher。完成分发器Completion Dispatcher收到通知后将通知传递给Proactor。此时Proactor将事件分发给相应的异步事件处理器Asynchronous Handlers。异步事件处理器Asynchronous Handlers收到分发的I/O事件后执行相应的操作如读取数据、处理业务逻辑、发送响应等。
从上述处理流程中我们可以发现 Proactor 模型的最大特点是采用异步 I/O。所有的 I/O 操作都由系统提供的异步 I/O 接口执行工作线程仅负责处理业务逻辑。在 Proactor 模型中用户函数启动一个异步文件操作并将其注册到多路复用器上。多路复用器关注的是异步读操作是否完成而不是文件是否可读或可写。异步操作由操作系统完成用户程序无需关心。当操作系统完成读文件操作即将数据复制到用户先前提供的缓冲区后便通知多路复用器相关操作已完成。多路复用器随后调用相应的处理程序处理数据。
尽管 Proactor 增加了编程的复杂度但它提高了工作线程的效率。Proactor 可在系统态优化读写操作利用 I/O 并行能力从而实现高性能单线程模型。在 Windows 上由于缺乏类似 epoll 的机制因此采用 IOCP 支持高并发。由于操作系统进行了良好的优化Windows 更常使用基于完成端口的 Proactor 模型实现服务器。而在 Linux 上虽然 2.6 内核引入了 aio 接口但实际效果并不理想。aio 的出现主要是为了解决 poll 性能不佳的问题但实际测试表明epoll 的性能高于 pollaio且 aio 无法处理 accept。因此Linux 主要还是以 Reactor 模型为主。
在不使用操作系统提供的异步 I/O 接口的情况下确实可以通过 Reactor 模拟 Proactor。区别在于利用异步接口可以使用系统提供的读写并行能力而在模拟情况下这需要在用户态实现。具体的做法包括以下步骤
注册读事件同时提供一段缓冲区。事件分离器等待可读事件。事件到来时激活分离器分离器立即读取数据并写入缓冲区然后调用事件处理器。事件处理器处理数据删除事件需要再用异步接口注册。
值得注意的是Boost.asio 库采用的是 Proactor 模型。然而在 Linux 平台上Boost.asio 使用 epoll 实现的 Reactor 来模拟 Proactor并额外开辟一个线程来完成读写调度。这种做法在一定程度上结合了两种模型的优势实现了较高的性能和灵活性。 在C里使用Proactor模型实现的TCP回显服务器
这个示例基于Boost.Asio库它是一个跨平台的C库用于编写基于异步I/O操作的网络和低级I/O程序使用前请确保你已经正确安装并配置好共享库文件。
#include iostream
#include memory
#include utility
#include boost/asio.hppusing boost::asio::ip::tcp;class session : public std::enable_shared_from_thissession {
public:session(tcp::socket socket) : socket_(std::move(socket)) {}void start() { read(); }private:void read() {auto self(shared_from_this());socket_.async_read_some(boost::asio::buffer(data_, max_length),[this, self](boost::system::error_code ec, std::size_t length) {if (!ec) {write(length);}});}void write(std::size_t length) {auto self(shared_from_this());boost::asio::async_write(socket_, boost::asio::buffer(data_, length),[this, self](boost::system::error_code ec, std::size_t /*length*/) {if (!ec) {read();}});}tcp::socket socket_;enum { max_length 1024 };char data_[max_length];
};class server {
public:server(boost::asio::io_context io_context, short port): acceptor_(io_context, tcp::endpoint(tcp::v4(), port)) {accept();}private:void accept() {acceptor_.async_accept([this](boost::system::error_code ec, tcp::socket socket) {if (!ec) {std::make_sharedsession(std::move(socket))-start();}accept();});}tcp::acceptor acceptor_;
};int main(int argc, char* argv[]) {try {if (argc ! 2) {std::cerr Usage: echo_server port\n;return 1;}boost::asio::io_context io_context;server s(io_context, std::atoi(argv[1]));io_context.run();} catch (std::exception e) {std::cerr Exception: e.what() \n;}return 0;
}使用telnet测试
3. 同步IO模拟Proactor模型 主线程将 socket 上的读就绪事件注册到 epoll 内核事件表中。主线程调用 epoll_wait 等待 socket 上的数据可读事件。当 socket 上有数据可读时epoll_wait 通知主线程。主线程循环读取 socket 上的数据直到无更多数据可读然后将读取到的数据封装为请求对象并插入请求队列。在请求队列上等待的某个工作线程被唤醒获取请求对象并处理客户请求然后在 epoll 内核事件表中注册 socket 上的写就绪事件。主线程调用 epoll_wait 等待 socket 的可写事件。当 socket 可写时epoll_wait 通知主线程。主线程将服务器处理客户请求的结果写入 socket。
两种模式的相同之处在于它们都涉及对某个 I/O 事件的通知即通知某个模块这个 I/O 操作可以进行或已经完成。在结构上两者也有共同点demultiplexor 负责提交 I/O 操作异步、查询设备是否可操作同步当条件满足时回调注册的处理函数。
它们的不同之处在于在 Proactor异步情况下回调注册的处理函数时表示 I/O 操作已经完成而在 Reactor同步情况下回调注册的处理函数时表示 I/O 设备可以进行某个操作可读或可写此时注册的处理函数开始提交操作。 在C中我们可以使用Boost.Asio库来使用步同步I/O模拟Proactor模型。以下是使用Boost.Asio实现Proactor模型的步骤
安装Boost库首先确保已经安装了Boost库并将其包含在项目中。Boost.Asio是Boost库的一部分。包含所需的头文件#include boost/asio.hpp
#include boost/bind.hpp
#include iostream
#include vector创建异步回调函数为了模拟Proactor模型我们需要创建异步回调函数该函数在异步操作完成时被调用。void handle_read(const boost::system::error_code error, std::size_t bytes_transferred) {if (!error) {std::cout Read: bytes_transferred bytes std::endl;} else {std::cerr Error: error.message() std::endl;}
}使用Boost.Asio创建异步操作使用Boost.Asio创建一个异步读取操作并将其与异步回调函数关联。boost::asio::io_context io_context;
boost::asio::ip::tcp::socket socket(io_context);
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(127.0.0.1), 8080);socket.async_connect(endpoint, boost::bind(handle_read, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));运行I/O上下文运行I/O上下文以处理异步操作。io_context.run();完整的代码如下
#include boost/asio.hpp
#include boost/bind.hpp
#include iostream
#include vectorvoid handle_read(const boost::system::error_code error, std::size_t bytes_transferred) {if (!error) {std::cout Read: bytes_transferred bytes std::endl;} else {std::cerr Error: error.message() std::endl;}
}int main() {try {boost::asio::io_context io_context;boost::asio::ip::tcp::socket socket(io_context);boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(127.0.0.1), 8080);socket.async_connect(endpoint, boost::bind(handle_read, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));io_context.run();} catch (std::exception e) {std::cerr Exception: e.what() std::endl;}return 0;
}测试使用的TCP服务器来接受上面客户端的连接
#include boost/asio.hpp
#include iostream
#include array
#include memoryclass session : public std::enable_shared_from_thissession {
public:session(boost::asio::ip::tcp::socket socket) : socket_(std::move(socket)) {}void start() { read(); }private:void read() {auto self(shared_from_this());socket_.async_read_some(boost::asio::buffer(data_),[this, self](boost::system::error_code ec, std::size_t length) {if (!ec) {read();}});}boost::asio::ip::tcp::socket socket_;std::arraychar, 1024 data_;
};class server {
public:server(boost::asio::io_context io_context, short port): acceptor_(io_context, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)) {accept();}private:void accept() {acceptor_.async_accept([this](boost::system::error_code ec, boost::asio::ip::tcp::socket socket) {if (!ec) {std::make_sharedsession(std::move(socket))-start();}accept();});}boost::asio::ip::tcp::acceptor acceptor_;
};int main() {try {boost::asio::io_context io_context;server srv(io_context, 8080);io_context.run();} catch (std::exception e) {std::cerr Exception: e.what() std::endl;}return 0;
}makefile
.PHONY:clean
all: client.cc server.ccg client.cc -o client -I/usr/local/include -L/usr/local/lib -lpthread -stdc11g server.cc -o server -I/usr/local/include -L/usr/local/lib -lpthread -stdc11
clean:rm -f client server
