网站建设的制度建设,seo一个月赚多少钱,安平网站建设培训,网页微信扫码登录在C很多框架中都有异步事件处理机制#xff0c;这导致我们在看源码时经常很疑惑#xff0c;难以理解#xff0c;而其中包含的编程套路可能是一些成熟的技术#xff0c;只是我们不熟悉#xff0c;比如WebRTC中类似于Qt的信号槽机制#xff0c;线程事件处理, 或者使用系统异… 在C很多框架中都有异步事件处理机制这导致我们在看源码时经常很疑惑难以理解而其中包含的编程套路可能是一些成熟的技术只是我们不熟悉比如WebRTC中类似于Qt的信号槽机制线程事件处理, 或者使用系统异步IO等等如果看不懂这些套路理解代码会很难本篇博客来尝使用用C线程池实现一种异步事件处理机制。
异步事件处理机制的基本实现 C可以使用std::future和std::promise来实现异步操作。然而为了实现一个异步事件绑定的框架我们需要更复杂的设计。下面是一个简单的例子说明如何实现一个异步事件处理器。 首先定义一个事件处理器类该类将接收并处理事件
class EventHandler {
public:virtual ~EventHandler() default;virtual void handleEvent(int eventID) 0;
};然后我们需要创建一个事件分发器它将异步地调用事件处理器
/*事件注册分发*/#pragma once#include EventHandler.hpp
#include map
#include thread
#include future
#include functional
#include memoryclass EventDispatcher {
public:// 注册事件处理器void registerHandler(int eventID, std::shared_ptrEventHandler handler) {handlers[eventID] handler;}// 异步事件分发函数void postEvent(int eventID) {auto it handlers.find(eventID);if (it ! handlers.end()) {std::thread eventThread(EventDispatcher::dispatchEvent, this, it-second, eventID);eventThread.detach();}}private:// 事件分发函数void dispatchEvent(std::shared_ptrEventHandler handler, int eventID) {handler-handleEvent(eventID);}private:std::mapint, std::shared_ptrEventHandler handlers; // 存储事件int 事件id, std::shared_ptrEventHandler 事件处理器
};在这个例子中EventDispatcher类的postEvent方法接收一个事件ID并在新线程中调用相应的事件处理器。这样做可以实现事件的异步处理。 然后你可以创建一个或多个处理器类比如下面的打印事件处理器PrintEventHandler 它实现EventHandler接口
/*具体的事件处理器*/#include EventHandler.hpp
#include iostreamusing namespace std;class PrintEventHandler : public EventHandler {
public:void handleEvent(int eventID) override {std::cout Handling event eventID std::endl;}
};然后再main函数中进行注册
/*C异步事件框架demo01*/#include iostream
#include memory
#include thread
#include chrono
#include EventDispatcher.hpp
#include PrintEventHandler.hppint main() {EventDispatcher dispatcher;std::shared_ptrEventHandler printHandler std::make_sharedPrintEventHandler();dispatcher.registerHandler(1, printHandler);dispatcher.postEvent(1);// Sleep main thread to let the event thread finish.std::this_thread::sleep_for(std::chrono::seconds(1));return 0;
}运行结果
Handling event 1
代码组织如下有兴趣的可以自行编写实现
cmake脚本
#[[编译方法cmake -S . -B buildcd buildmake./demo01]]cmake_minimum_required(VERSION 3.20)project(demo01)set(INCLUDE_PATH1 ./)# 添加头文件目录
include_directories(${INCLUDE_PATH1}
)# 添加子目录src
aux_source_directory(./ SRC)add_executable(demo01 ${SRC})这个实现是非常基础的并没有考虑到线程安全问题和异常处理等等。在实际的项目中你需要更复杂的设计并使用更高级的并发编程技术如线程池、任务队列、互斥锁等等。
添加线程池、任务队列 如果想要更复杂的设计包括线程池、任务队列、互斥锁等你可以考虑使用以下的设计。下面的例子使用了C17的std::async和std::future来实现线程池和任务队列。 首先我们需要一个线程安全的任务队列
#pragma once#include queue
#include mutex
#include condition_variabletemplate typename T
class ThreadSafeQueue {
public:ThreadSafeQueue() default;ThreadSafeQueue(const ThreadSafeQueueT ) delete;ThreadSafeQueue operator(const ThreadSafeQueueT ) delete;void push(T value) {std::lock_guardstd::mutex lock(mMutex);mQueue.push(std::move(value));mCondition.notify_one();}bool try_pop(T value) {std::lock_guardstd::mutex lock(mMutex);if (mQueue.empty()) {return false;}value std::move(mQueue.front());mQueue.pop();return true;}void wait_and_pop(T value) {std::unique_lockstd::mutex lock(mMutex);mCondition.wait(lock, [this](){ return !mQueue.empty(); });value std::move(mQueue.front());mQueue.pop();}private:std::queueT mQueue;std::mutex mMutex;std::condition_variable mCondition;
};然后我们需要一个线程池来处理这些任务
#pragma once#include ThreadSafeQueue.hpp
#include vector
#include futureclass ThreadPool {
public:ThreadPool(size_t numThreads) {start(numThreads);}~ThreadPool() {stop();}templatetypename Tvoid enqueue(T task) {mTasks.push(std::make_sharedstd::packaged_taskvoid()(std::move(task)));}private:std::vectorstd::thread mThreads;ThreadSafeQueuestd::shared_ptrstd::packaged_taskvoid() mTasks;std::atomicbool mContinue { true };void start(size_t numThreads) {for (auto i 0u; i numThreads; i) {mThreads.emplace_back([] {while (mContinue) {std::shared_ptrstd::packaged_taskvoid() task;if (mTasks.try_pop(task)) {(*task)();} else {std::this_thread::yield();}}});}}void stop() noexcept {mContinue false;for (auto thread : mThreads) {thread.join();}}
};然后我们可以在EventDispatcher中使用线程池来异步地处理事件
#pragma once
#include ThreadPool.hpp
#include mutex
#include map
#include memory
#include EventHandler.hppclass EventDispatcher {
public:EventDispatcher(size_t numThreads) : mThreadPool(numThreads) {}void registerHandler(int eventID, std::shared_ptrEventHandler handler) {std::lock_guardstd::mutex lock(mMutex);mHandlers[eventID] handler;}void postEvent(int eventID) {std::lock_guardstd::mutex lock(mMutex);auto it mHandlers.find(eventID);if (it ! mHandlers.end()) {mThreadPool.enqueue([handler it-second, eventID]() {handler-handleEvent(eventID);});}}private:std::mapint, std::shared_ptrEventHandler mHandlers;ThreadPool mThreadPool;std::mutex mMutex;
};这个实现现在是线程安全的并且使用了线程池来处理事件。对于异常处理你可以在EventHandler的handleEvent方法内部进行。
添加异常处理 如何在EventDispatcher中处理异常 在EventDispatcher中处理异常我们可以在任务的执行过程中增加一个try/catch块捕获并处理可能出现的异常。下面是一个示例
class EventDispatcher {
public:EventDispatcher(size_t numThreads) : mThreadPool(numThreads) {}void registerHandler(int eventID, std::shared_ptrEventHandler handler) {std::lock_guardstd::mutex lock(mMutex);mHandlers[eventID] handler;}void postEvent(int eventID) {std::lock_guardstd::mutex lock(mMutex);auto it mHandlers.find(eventID);if (it ! mHandlers.end()) {mThreadPool.enqueue([handler it-second, eventID]() {try {handler-handleEvent(eventID);} catch (const std::exception ex) {std::cerr Exception occurred while handling event eventID : ex.what() \n;} catch (...) {std::cerr Unknown exception occurred while handling event eventID \n;}});}}private:std::mapint, std::shared_ptrEventHandler mHandlers;ThreadPool mThreadPool;std::mutex mMutex;
};在这个示例中如果handleEvent方法抛出了一个异常那么我们将在控制台上打印出一个错误消息。具体如何处理这些异常取决于你的应用程序的特定需求。 请注意这种方法仅处理了在handleEvent方法中抛出的异常。如果在其他地方如ThreadPool的构造函数或enqueue方法抛出异常你可能需要额外的代码来处理这些情况。
注册多个事件 在我们的事件分发器设计中注册多个事件是非常简单的。你只需要创建不同的事件处理器然后使用不同的事件ID将它们注册到事件分发器中。以下是一个示例
class PrintEventHandler : public EventHandler {
public:PrintEventHandler(int id) : id(id) {}void handleEvent(int eventID) override {std::cout Handler id handling event eventID std::endl;}
private:int id;
};int main() {EventDispatcher dispatcher(4); // 创建一个有4个线程的事件分发器std::shared_ptrEventHandler handler1 std::make_sharedPrintEventHandler(1);dispatcher.registerHandler(1, handler1);std::shared_ptrEventHandler handler2 std::make_sharedPrintEventHandler(2);dispatcher.registerHandler(2, handler2);dispatcher.postEvent(1);dispatcher.postEvent(2);// Sleep main thread to let the event threads finishstd::this_thread::sleep_for(std::chrono::seconds(1));return 0;
}在这个示例中我们创建了两个PrintEventHandler对象每个对象都有一个唯一的ID。然后我们使用不同的事件ID将这两个处理器注册到EventDispatcher中。当我们发布事件时相应的处理器将会被调用。 这就是如何在我们的异步事件绑定框架中注册多个事件的方法。
如何取消注册 要从事件分发器中取消注册一个事件处理器你可以在EventDispatcher类中添加一个新的方法例如unregisterHandler用于从处理器映射中删除指定的事件ID。下面是如何实现这个函数的例子
class EventDispatcher {
public:EventDispatcher(size_t numThreads) : mThreadPool(numThreads) {}void registerHandler(int eventID, std::shared_ptrEventHandler handler) {std::lock_guardstd::mutex lock(mMutex);mHandlers[eventID] handler;}void unregisterHandler(int eventID) {std::lock_guardstd::mutex lock(mMutex);mHandlers.erase(eventID);}void postEvent(int eventID) {std::lock_guardstd::mutex lock(mMutex);auto it mHandlers.find(eventID);if (it ! mHandlers.end()) {mThreadPool.enqueue([handler it-second, eventID]() {try {handler-handleEvent(eventID);} catch (const std::exception ex) {std::cerr Exception occurred while handling event eventID : ex.what() \n;} catch (...) {std::cerr Unknown exception occurred while handling event eventID \n;}});}}private:std::mapint, std::shared_ptrEventHandler mHandlers;ThreadPool mThreadPool;std::mutex mMutex;
};这个unregisterHandler函数将删除与给定事件ID关联的事件处理器。请注意这个函数并不会停止已经在处理该事件的任何线程。如果你想要取消正在进行的事件处理你可能需要一个更复杂的设计例如使用std::future和std::promise来控制线程的执行。
如何停止正在进行的事件 要停止正在进行的事件处理我们需要更复杂的设计它可能包括使用std::future和std::promise来控制线程的执行。在这种设计中每当一个事件被发布时我们将创建一个std::promise并将相应的std::future存储在某个地方以便我们可以稍后在需要时停止事件处理。 但是要注意的是根据C的设计没有一个简单且安全的方法可以强制停止正在运行的线程因为这可能会导致资源泄漏或其他未定义的行为。因此更常见的做法是让事件处理器定期检查一个“停止标记”然后在检查到该标记时优雅地停止执行。以下是一个简单的示例演示了如何实现这种设计
class StoppableEvent {
public:StoppableEvent(std::futurevoid future, std::functionvoid() func): mFuture(std::move(future)), mFunc(std::move(func)) {}void operator()() {while(mFuture.wait_for(std::chrono::milliseconds(100)) std::future_status::timeout) {mFunc();}}private:std::futurevoid mFuture;std::functionvoid() mFunc;
};class EventDispatcher {
public:EventDispatcher(size_t numThreads) : mThreadPool(numThreads) {}void registerHandler(int eventID, std::shared_ptrEventHandler handler) {std::lock_guardstd::mutex lock(mMutex);mHandlers[eventID] handler;}void postEvent(int eventID) {std::lock_guardstd::mutex lock(mMutex);auto it mHandlers.find(eventID);if (it ! mHandlers.end()) {std::promisevoid stopSignal;auto stopFuture stopSignal.get_future();mStopSignals[eventID] std::move(stopSignal);mThreadPool.enqueue(StoppableEvent(std::move(stopFuture), [handler it-second, eventID]() {handler-handleEvent(eventID);}));}}void stopEvent(int eventID) {std::lock_guardstd::mutex lock(mMutex);auto it mStopSignals.find(eventID);if (it ! mStopSignals.end()) {it-second.set_value();mStopSignals.erase(it);}}private:std::mapint, std::shared_ptrEventHandler mHandlers;std::mapint, std::promisevoid mStopSignals;ThreadPool mThreadPool;std::mutex mMutex;
};在这个例子中我们定义了一个StoppableEvent类它将一个std::future和一个函数组合在一起。当operator()被调用时它将定期检查future如果future的状态不是timeout则停止执行函数。 然后当我们在EventDispatcher中发布一个事件时我们将创建一个新的std::promise和相应的std::future并将这个future和事件处理器的handleEvent方法一起传递给StoppableEvent。我们还将promise存储在一个映射中以便我们可以稍后通过调用set_value来发出停止信号。 最后我们添加了一个stopEvent方法它将查找与给定事件ID关联的promise并通过调用set_value来发出停止信号。然后它将从映射中删除这个promise因为我们不再需要它。 这是一个基本的示例你可能需要根据你的具体需求来修改和扩展它。请注意这个设计假设事件处理器的handleEvent方法将被调用多次每次调用都可能被中断。如果你的事件处理器只执行一次长时间运行的任务那么这个设计可能并不适合。 以上是一个简易的异步事件处理demo, 在项目开发中需要根据具体的业务需求进行调整完善。
