做网站的背景怎么做,足球网站模板下载,品牌策划是什么,做社交网站首先#xff0c;要搞懂一点#xff0c;异步操作本质上也是并发#xff0c;我们想要在线程级别实现异步并发基本就靠三种方式#xff1a;
多线程并发回调函数协程
今天我们讨论的是回调函数#xff0c;我们如何通过回调函数来实现异步操作呢#xff1f;
非阻塞I/O操作回…首先要搞懂一点异步操作本质上也是并发我们想要在线程级别实现异步并发基本就靠三种方式
多线程并发回调函数协程
今天我们讨论的是回调函数我们如何通过回调函数来实现异步操作呢
非阻塞I/O操作回调函数实现异步IO基于定时器回调函数实现异步任务调度事件队列回调函数
我们就分别来实现一下他们吧代码比较长请耐心阅读。
非阻塞I/O回调
这种方式允许程序在等待 I/O 操作完成时继续执行其他任务从而提高并发性和性能。
我们可以使用标准 C 库和 C11 中的线程库来实现一个基于 epoll 的非阻塞 I/O 和回调函数的示例。epoll 是 Linux 提供的高效 I/O 多路复用机制适用于处理大量并发连接。我们将使用 epoll 来实现异步 I/O并使用回调函数处理完成的 I/O 操作。
实例代码使用 epoll 实现非阻塞 I/O 和回调函数
#include sys/epoll.h
#include unistd.h
#include fcntl.h
#include iostream
#include cstring
#include functional
#include unordered_map
#include vector
#include thread
#include chronovoid set_non_blocking(int fd) {int flags fcntl(fd, F_GETFL, 0);if (flage -1) {throw std::runtime_error(fcntl get error);}if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) -1) {throw std::runtime_error(fcntl set error);}
}//事件处理器类
class EventLoop {
public:EventLoop() {epollFd_ epoll_create1(0);if (epoll_fd -1) {throw std::runtime_error(epoll_creat1 error);}}~EpollLoop() {close(epoll_fd);}//添加文件描述符及其对应的回调函数void add_fd(int fd, std::functionvoid(int) cb) {set_non_blocking(fd);epoll_event event;event.events EPOLLIN | EPOLLET;event.data.fd fd;if (epoll_ctl(epollFd_, EPOLL_CTL_ADD, fd, event) -1) {throw std::runtime_error(epoll_ctl add error);}callbacks_[fd] cb;}//运行时间循环void run() {while (running_) {std::vectorepoll_event events(10);int numActives epoll_wait(epollFd_, events.data(), events.size(), -1);if (numActives -1) {throw std::runtime_error(epoll_wait error);}for (int i 0; i n; i) {int fd events[i].data.fd;if (callbacks_.find(fd) ! callbacks_.end()) callbacks[fd](fd);}}}//停止事件循环void stop() {return false;}private:int epollFd_; //epoll实例std::unordered_mapint, std::functionvoid(int) callbacks_;bool running_ true;
};// 示例回调函数处理标准输入的读取
void handle_stdin(int fd) {char buffer[128];ssize_t n read(fd, buffer, sizeof(buffer) - 1);if (n 0) {buffer[n] \0;std::cout Read from stdin: buffer std::endl;} else if (n 0) {std::cout EOF from stdin std::endl;} else {if (errno ! EAGAIN errno ! EWOULDBLOCK) {std::ceer Read error: strerror(errno) std::endl;} }
}int main () {try {EventLoop loop;//注册标准输入的文件描述符和回调函数loop.add_fd(STDIN_FILENO, handle_stdin);//运行事件循环std::thread loop_thread([loop]() {loop.run();});//模拟只线程的其他工作std::this_thread::sleep_for(std::chrono::seconds(10));//停止事件循环loop.stop();loop_thread.join();} catch (const std::exception e) {std::cerr Exception: e.what() std::endl;return 1}return 0;
}在该示例中由于是非阻塞IO所以如果有两个 fd 对应的事件被激活那么他们会循环执行各自的回调操作因为我们的EventLoop里面是又一个 while 循环的如果这两个 fd 如果都是读一个很大很大的文件那么他们在while循环中会交替执行各自的回调任务实现异步操作。
基于定时器回调函数实现异步任务调度
通过定时器和回调函数可以在单线程环境中实现异步任务调度。在这个示例中我们实现了一个简单的 Web 服务器定时清理过期会话的功能。类似的技术可以应用于其他需要定时任务调度的场景如定时备份、日志轮换、定时数据采集等。
1. SessionManager 类
这个类负责管理用户会话提供添加、删除和清理过期会话的方法。
class SessionManager {
#include iostream
#include unordered_map
#include chrono
#include functional
#include thread
#include vector
#include algorithmclass SessionManager {
public:void add_session(const std::string session_id) {sessions[session_id] std::chrono::steady_clock::now();}void remove_session(const std::string session_id) {sessions.erase(session_id);}void clean_expired_sessions(std::chrono::seconds timeout) {auto now std::chrono::steady_clock::now();for (auto it sessions.begin(); it ! sessions.end(); ) {if (now - it-second timeout) {std::cout Removing expired session: it-first std::endl;it sessions.erase(it);} else {it;}}}private:std::unordered_mapstd::string, std::chrono::steady_clock::time_point sessions;
};add_session添加一个新会话记录会话 ID 和当前时间。remove_session删除指定的会话。clean_expired_sessions清理过期会话检查所有会话如果会话时间超过指定的超时时间则删除会话。
2. EventLoop类
这个类管理定时任务通过维护一个事件列表并运行事件循环在指定时间间隔内执行任务。
class EventLoop {
public:void add_event(std::functionvoid() callback, std::chrono::milliseconds interval) {auto next_run std::chrono::steady_clock::now() interval;events.emplace_back(next_run, interval, callback);}void run() {while (running) {auto now std::chrono::steady_clock::now();for (auto event : events) {if (now event.next_run) {event.callback();event.next_run now event.interval;}}std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 小休眠减少CPU占用}}void stop() {running false;}private:struct Event {std::chrono::steady_clock::time_point next_run;std::chrono::milliseconds interval;std::functionvoid() callback;Event(std::chrono::steady_clock::time_point nr, std::chrono::milliseconds i, std::functionvoid() cb): next_run(nr), interval(i), callback(cb) {}};std::vectorEvent events;bool running true;
};add_event添加一个新的定时事件指定回调函数和时间间隔。run启动事件循环循环检查所有事件如果事件到期则执行回调函数并更新下一次执行时间。stop停止事件循环。
3. 回调函数
定义一个回调函数用于清理过期会话。 如果我们定义两个不同任务的回调函数那么是不是就可以在某段时间实现多个任务了呢
void clean_sessions_task(SessionManager session_manager, std::chrono::seconds timeout) {session_manager.clean_expired_sessions(timeout);
}这个函数调用 SessionManager 的 clean_expired_sessions 方法来清理过期会话。
4. main 函数
int main() {SessionManager session_manager;EventLoop event_loop;// 添加一些示例会话session_manager.add_session(session1);std::this_thread::sleep_for(std::chrono::seconds(1));session_manager.add_session(session2);// 每2秒清理一次过期会话假设过期时间为3秒event_loop.add_event(std::bind(clean_sessions_task, std::ref(session_manager), std::chrono::seconds(3)), std::chrono::seconds(2));// 运行事件循环std::thread event_loop_thread([event_loop]() {event_loop.run();});// 模拟服务器运行std::this_thread::sleep_for(std::chrono::seconds(10));// 停止事件循环event_loop.stop();event_loop_thread.join();return 0;
}创建 SessionManager 和 EventLoop 对象。添加两个示例会话分别在 0 秒和 1 秒时添加。添加一个定时事件每 2 秒调用一次 clean_sessions_task 来清理过期会话过期时间为 3 秒。启动一个线程运行事件循环。主线程模拟服务器运行 10 秒钟。停止事件循环并等待事件循环线程结束。
如果我们添加两个定时事件并且打开循环那么不是就在某段时间实现了异步任务调度呢
基于事件队列回调函数实现异步操作
这个示例展示了如何使用标准 C 库和 C11 线程库通过事件队列和回调函数在单线程中实现异步任务调度。具体来说我们实现了一个简单的系统可以调度并执行延迟任务。
1.EventLoop 类
这个类管理定时任务通过维护一个事件列表并运行事件循环在指定时间间隔内执行任务。
class EventLoop {
public:// 添加一个新的定时事件void add_event(std::functionvoid() callback, std::chrono::milliseconds interval) {auto next_run std::chrono::steady_clock::now() interval;events.emplace_back(next_run, interval, callback);}// 运行事件循环void run() {while (running) {auto now std::chrono::steady_clock::now();for (auto event : events) {if (now event.next_run) {event.callback(); // 执行回调函数event.next_run now event.interval; // 更新下一次执行时间}}std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 小休眠减少CPU占用}}// 停止事件循环void stop() {running false;}private:struct Event {std::chrono::steady_clock::time_point next_run;std::chrono::milliseconds interval;std::functionvoid() callback;Event(std::chrono::steady_clock::time_point nr, std::chrono::milliseconds i, std::functionvoid() cb): next_run(nr), interval(i), callback(cb) {}};std::vectorEvent events;bool running true;
};add_event添加一个新的定时事件指定回调函数和时间间隔。 参数 callback 是一个回调函数当事件触发时调用。参数 interval 是一个时间间隔表示事件应该在多长时间后执行。next_run 记录了事件的下一次执行时间。 run启动事件循环循环检查所有事件如果事件到期则执行回调函数并更新下一次执行时间。 while (running)事件循环在 running 为 true 时持续运行。event.callback()执行回调函数表示事件触发。event.next_run now event.interval更新事件的下一次执行时间。std::this_thread::sleep_for(std::chrono::milliseconds(10))通过短暂休眠来减少 CPU 占用。 stop停止事件循环将 running 设置为 false。
2. 回调函数
定义两个简单的回调函数用于示例任务。
void say_hello() {std::cout Hello, World! std::endl;
}void say_goodbye() {std::cout Goodbye, World! std::endl;
}
3.main函数
设置并运行事件循环添加一些示例任务并定期执行这些任务。
int main() {EventLoop loop;// 添加定时事件loop.add_event(say_hello, std::chrono::seconds(1)); // 每1秒执行一次loop.add_event(say_goodbye, std::chrono::seconds(2)); // 每2秒执行一次// 运行事件循环loop.run();return 0;
}异步任务的实现
异步任务的实现核心在于 EventLoop 类中的 add_event 和 run 方法
add_event 方法将任务回调函数和执行时间间隔添加到事件队列中。run 方法启动事件循环定期检查事件队列触发到期的任务并通过回调函数执行这些任务。
具体来说异步任务的执行发生在以下代码行
if (now event.next_run) {event.callback(); // 执行回调函数event.next_run now event.interval; // 更新下一次执行时间
}event.callback()当当前时间 now 大于等于 event.next_run 时执行回调函数表示异步任务触发并执行。
这个机制允许在单线程环境中通过事件队列和回调函数实现异步任务调度无需多线程或协程。每个任务在指定的时间间隔后触发保持事件循环的运行和任务的调度。
事件队列\定时器 回调真的能实现异步任务吗
当回调任务是一个非常耗时的操作时如果在单线程中执行确实会导致事件循环被阻塞从而影响其他任务的执行。这违背了异步执行的初衷即不阻塞程序的其他部分。
为了解决这个问题主要还是以下方法
使用线程池 我们可以使用一个线程池来处理耗时的任务从而避免阻塞事件循环。线程池可以并发执行多个任务确保事件循环保持响应。
