C++使用线程池模拟异步事件处理机制

news2024/12/24 9:41:21

  在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 <memory>

class EventDispatcher {
public:
    // 注册事件处理器
    void registerHandler(int eventID, std::shared_ptr<EventHandler> 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_ptr<EventHandler> handler, int eventID) {
        handler->handleEvent(eventID);
    }

private:
    std::map<int, std::shared_ptr<EventHandler>> handlers;  // 存储事件,int 事件id, std::shared_ptr<EventHandler> 事件处理器
};

  在这个例子中,EventDispatcher类的postEvent方法接收一个事件ID,并在新线程中调用相应的事件处理器。这样做可以实现事件的异步处理。

  然后,你可以创建一个或多个处理器类,比如下面的打印事件处理器PrintEventHandler ,它实现EventHandler接口,

/*

具体的事件处理器

*/

#include "EventHandler.hpp"
#include <iostream>

using 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.hpp"

int main() {
    EventDispatcher dispatcher;
    
    std::shared_ptr<EventHandler> printHandler = std::make_shared<PrintEventHandler>();
    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 build
    cd build
    make
    ./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})

  这个实现是非常基础的,并没有考虑到线程安全问题和异常处理等等。在实际的项目中,你需要更复杂的设计,并使用更高级的并发编程技术,如线程池、任务队列、互斥锁等等。

添加线程池、任务队列

  如果想要更复杂的设计,包括线程池、任务队列、互斥锁等,你可以考虑使用以下的设计。下面的例子使用了C++17的std::async和std::future来实现线程池和任务队列。

  首先,我们需要一个线程安全的任务队列:

#pragma once

#include <queue>
#include <mutex>
#include <condition_variable>

template <typename T>
class ThreadSafeQueue {
public:
    ThreadSafeQueue() = default;
    ThreadSafeQueue(const ThreadSafeQueue<T> &) = delete;
    ThreadSafeQueue& operator=(const ThreadSafeQueue<T> &) = delete;

    void push(T value) {
        std::lock_guard<std::mutex> lock(mMutex);
        mQueue.push(std::move(value));
        mCondition.notify_one();
    }

    bool try_pop(T& value) {
        std::lock_guard<std::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_lock<std::mutex> lock(mMutex);
        mCondition.wait(lock, [this](){ return !mQueue.empty(); });
        value = std::move(mQueue.front());
        mQueue.pop();
    }

private:
    std::queue<T> mQueue;
    std::mutex mMutex;
    std::condition_variable mCondition;
};

  然后,我们需要一个线程池来处理这些任务:

#pragma once

#include "ThreadSafeQueue.hpp"
#include <vector>
#include <future>

class ThreadPool {
public:
    ThreadPool(size_t numThreads) {
        start(numThreads);
    }

    ~ThreadPool() {
        stop();
    }

    template<typename T>
    void enqueue(T task) {
        mTasks.push(std::make_shared<std::packaged_task<void()>>(std::move(task)));
    }

private:
    std::vector<std::thread> mThreads;
    ThreadSafeQueue<std::shared_ptr<std::packaged_task<void()>>> mTasks;
    std::atomic<bool> mContinue { true };

    void start(size_t numThreads) {
        for (auto i = 0u; i < numThreads; ++i) {
            mThreads.emplace_back([=] {
                while (mContinue) {
                    std::shared_ptr<std::packaged_task<void()>> 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.hpp"

class EventDispatcher {
public:
    EventDispatcher(size_t numThreads) : mThreadPool(numThreads) {}

    void registerHandler(int eventID, std::shared_ptr<EventHandler> handler) {
        std::lock_guard<std::mutex> lock(mMutex);
        mHandlers[eventID] = handler;
    }

    void postEvent(int eventID) {
        std::lock_guard<std::mutex> lock(mMutex);
        auto it = mHandlers.find(eventID);
        if (it != mHandlers.end()) {
            mThreadPool.enqueue([handler = it->second, eventID]() {
                handler->handleEvent(eventID);
            });
        }
    }

private:
    std::map<int, std::shared_ptr<EventHandler>> 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_ptr<EventHandler> handler) {
        std::lock_guard<std::mutex> lock(mMutex);
        mHandlers[eventID] = handler;
    }

    void postEvent(int eventID) {
        std::lock_guard<std::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::map<int, std::shared_ptr<EventHandler>> 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_ptr<EventHandler> handler1 = std::make_shared<PrintEventHandler>(1);
    dispatcher.registerHandler(1, handler1);

    std::shared_ptr<EventHandler> handler2 = std::make_shared<PrintEventHandler>(2);
    dispatcher.registerHandler(2, handler2);

    dispatcher.postEvent(1);
    dispatcher.postEvent(2);

    // Sleep main thread to let the event threads finish
    std::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_ptr<EventHandler> handler) {
        std::lock_guard<std::mutex> lock(mMutex);
        mHandlers[eventID] = handler;
    }

    void unregisterHandler(int eventID) {
        std::lock_guard<std::mutex> lock(mMutex);
        mHandlers.erase(eventID);
    }

    void postEvent(int eventID) {
        std::lock_guard<std::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::map<int, std::shared_ptr<EventHandler>> mHandlers;
    ThreadPool mThreadPool;
    std::mutex mMutex;
};

  这个unregisterHandler函数将删除与给定事件ID关联的事件处理器。请注意,这个函数并不会停止已经在处理该事件的任何线程。如果你想要取消正在进行的事件处理,你可能需要一个更复杂的设计,例如使用std::futurestd::promise来控制线程的执行。

如何停止正在进行的事件

  要停止正在进行的事件处理,我们需要更复杂的设计,它可能包括使用std::futurestd::promise来控制线程的执行。在这种设计中,每当一个事件被发布时,我们将创建一个std::promise,并将相应的std::future存储在某个地方,以便我们可以稍后在需要时停止事件处理。

  但是,要注意的是,根据C++的设计,没有一个简单且安全的方法可以强制停止正在运行的线程,因为这可能会导致资源泄漏或其他未定义的行为。因此,更常见的做法是让事件处理器定期检查一个“停止标记”,然后在检查到该标记时优雅地停止执行。以下是一个简单的示例,演示了如何实现这种设计:

class StoppableEvent {
public:
    StoppableEvent(std::future<void> future, std::function<void()> 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::future<void> mFuture;
    std::function<void()> mFunc;
};

class EventDispatcher {
public:
    EventDispatcher(size_t numThreads) : mThreadPool(numThreads) {}

    void registerHandler(int eventID, std::shared_ptr<EventHandler> handler) {
        std::lock_guard<std::mutex> lock(mMutex);
        mHandlers[eventID] = handler;
    }

    void postEvent(int eventID) {
        std::lock_guard<std::mutex> lock(mMutex);
        auto it = mHandlers.find(eventID);
        if (it != mHandlers.end()) {
            std::promise<void> 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_guard<std::mutex> lock(mMutex);
        auto it = mStopSignals.find(eventID);
        if (it != mStopSignals.end()) {
            it->second.set_value();
            mStopSignals.erase(it);
        }
    }

private:
    std::map<int, std::shared_ptr<EventHandler>> mHandlers;
    std::map<int, std::promise<void>> 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, 在项目开发中,需要根据具体的业务需求进行调整完善。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1196322.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

玩转Linux基本指令

> 作者简介&#xff1a;დ旧言~&#xff0c;目前大二&#xff0c;现在学习Java&#xff0c;c&#xff0c;c&#xff0c;Python等 > 座右铭&#xff1a;松树千年终是朽&#xff0c;槿花一日自为荣。 > 目标&#xff1a;牢记Linux的基本指令。 > 毒鸡汤&#xff1a;挫…

2023 年最好的 Android 系统修复/刷机应用程序和软件

任何 Android 设备要顺利运行&#xff0c;其操作系统必须运行良好。幸运的是&#xff0c;对于大多数 Android 用户来说&#xff0c;这是不间断的。设备运行良好&#xff0c;打电话、共享文档等都没有问题。尽管如此&#xff0c;Android 操作系统可能会停止运行。这可能是由于特…

Postman for Mac(HTTP请求发送调试工具)v10.18.10官方版

Postman for mac是一个提供在MAC设备上功能强大的开发&#xff0c;监控和测试API的绝佳工具。非常适合开发人员去使用。此版本通过Interceptor添加了对请求捕获的支持&#xff0c;修正了使用上下文菜单操作未复制响应正文的问题和预请求脚本的垂直滚动条与自动完成下拉列表重叠…

ios开发 之 多线程

目录 第一节&#xff1a;多线程简介 线程执行原理 主线程 多线程解决方案 pthread __bridge NSThread 线程的状态 第二节&#xff1a;多线程访问资源 Synchronized nonatomic 、atomic 自动释放池 属性修饰符 第三节&#xff1a;消息循环 消息模式 第四节&…

人工智能基础_机器学习022_使用正则化_曼哈顿距离_欧氏距离_提高模型鲁棒性_过拟合_欠拟合_正则化提高模型泛化能力---人工智能工作笔记0062

然后我们再来看一下,过拟合和欠拟合,现在,实际上欠拟合,出现的情况已经不多了,欠拟合是 在训练集和测试集的准确率不高,学习不到位的情况. 然后现在一般碰到的是过拟合,可以看到第二个就是,完全就把红点蓝点分开了,这种情况是不好的, 因为分开是对训练数据进行分开的,如果来…

三掌柜第2期赠书活动:《计算机考研精炼1000题》

引言 各位朋友大家好&#xff0c;我是三掌柜。今天&#xff0c;三掌柜赠书第2期启动&#xff0c;本次为大家精选了《计算机考研精炼1000题》这本书。关于这本书的内容&#xff0c;非常丰富&#xff0c;涵盖计算机考研的高频知识内容&#xff0c;不管是正在备考&#xff0c;还是…

JUC下常见类

JUC(java.util.concurrent) 的常见类ReentrantLock原子类线程池信号量SemaphoreCountDownLatch JUC(java.util.concurrent) 的常见类 ReentrantLock ReentrantLock可重入互斥锁. 和 synchronized 定位类似, 都是用来实现互斥效果, 保证线程安全。 用法: lock(): 加锁, 如果获…

合同审查---财务条款、合同形式与生效审查

1.合同主体 1人 廖 2.财务条款、合同形式与生效 1人 黄 3.履行、验收、知识产权、不可抗力 1人 詹 4.违约责任、争议解决、保密、法律引用 1人 王 代码规范&#xff1a; 1.代码函数的层级 各审查点在json中分为3级层级&#xff0c;但用python写规则的时候&#xff0c;1级层级为…

【C++那些事儿】类与对象(2)

君兮_的个人主页 即使走的再远&#xff0c;也勿忘启程时的初心 C/C 游戏开发 Hello,米娜桑们&#xff0c;这里是君兮_&#xff0c;我之前看过一套书叫做《明朝那些事儿》&#xff0c;把本来枯燥的历史讲的生动有趣。而C作为一门接近底层的语言&#xff0c;无疑是抽象且难度颇…

LeetCode 面试题 16.20. T9键盘

文章目录 一、题目二、C# 题解 一、题目 在老式手机上&#xff0c;用户通过数字键盘输入&#xff0c;手机将提供与这些数字相匹配的单词列表。每个数字映射到0至4个字母。给定一个数字序列&#xff0c;实现一个算法来返回匹配单词的列表。你会得到一张含有有效单词的列表。映射…

图书销售数据大屏可视化【可视化项目案例-03】

🎉🎊🎉 你的技术旅程将在这里启航! 🚀🚀 本文选自专栏:可视化技术专栏100例 可视化技术专栏100例,包括但不限于大屏可视化、图表可视化等等。订阅专栏用户在文章底部可下载对应案例源码以供大家深入的学习研究。 🎓 每一个案例都会提供完整代码和详细的讲解,不…

Leetcode-144 二叉树的前序遍历

递归方法 /*** Definition for a binary tree node.* public class TreeNode {* int val;* TreeNode left;* TreeNode right;* TreeNode() {}* TreeNode(int val) { this.val val; }* TreeNode(int val, TreeNode left, TreeNode right) {* …

谈谈如何写作(一)

序言 没有什么比一套好理论更有用了。——库尔特勒温 谈谈如何写作&#xff1f; 这个题目就像一本书《如何阅读一本书》&#xff0c;听起来似乎很无趣&#xff0c;因为各位盆友直接的反应&#xff0c;可能是这根本无需理论。 实际情况呢&#xff1f;恰恰相反&#xff0c;往往你…

javax.management.InstanceNotFoundException: Catalina:type=Server错误的解决

软件&#xff1a; JDK 1.8 Tomcat 8.5.66 IDEA 2019.3.3 问题&#xff1a;启动IDEA新建一Web Application项目&#xff0c;设置好项目运行&#xff0c;结果发现提示&#xff1a; 提示&#xff1a;Application Server was not connected before run configuration stop, rea…

JavaSE 类与对象

前言 我们之前学的都是面向过程&#xff0c;面向过程研究的是对单个对象的一种方法实现过程&#xff0c;比如求一个数的阶乘&#xff0c;强调的是怎么实现这个方法的过程&#xff0c;但对我们以后来说&#xff0c;如果想要应用到更广的层面&#xff0c;不能只是学习一个方法的…

基于Qt Linux开发板USER-KEY按键实现

介绍如何在 Qt 应用上使用嵌入式 GET6818 Linux 开发板 上的按键。 工具:Qt Creator 5.14.2 平台:windows ## 资源简介 在GET6818 开发板,开发板板载资源上有两个用户按键。如下图原理图(下图开发板的按键原理图)。 ## 应用实例 想要监测这个 KEY0,首先出厂内核已经…

Linux是什么,Linux系统介绍

很多小伙伴都不是那么了解和知道Linux&#xff0c;到底Linux是什么&#xff1f; 像大家用到的安卓手机&#xff0c;生活中用到的各种智能设备&#xff0c;比如路由器&#xff0c;光猫&#xff0c;智能家具等&#xff0c;很多都是在Linux操作系统上。 Linux是什么&#xff1f;Li…

常见问题: (Windows/app/浏览器)总结及其研究———(不断更新中.....)

问题目录 手机电脑电脑qq如何多开分身电脑与手机无线传送数据的方法 浏览器下载如何利用技术下载网上图片 WindowsVMware Workstation1 无法创建11264MB的匿名分页文件&#xff1a;页面文件2 虚拟机安装Windows11时出现: tempting to start up from: EFI VMware Virtual N 百度…

Vuex:模块化Module :VCA模式

VCA中不支持辅助函数&#xff0c;因为辅助函数中是用this.$store&#xff0c;而VCA中没有绑定this的 由于使用单一状态树&#xff0c;应用的所有状态会集中到一个比较大的对象。当应用变得非常复杂时&#xff0c;store 对象就有可能变得相当臃肿。 这句话的意思是&#xff0c;…

开源Gimp动态压感笔刷设置方法

一、问题描述 开源绘画工具的Gimp的笔刷压感在哪里控制和开启呢&#xff1f; 二、解决方法 1、Gimp有专用的笔刷集&#xff1a;如下图。开启需要在主窗口window下拉菜单开启&#xff0c;或在右侧面板里的左箭头按钮里打开。一般绘画够用了。比用自定义特殊笔刷。 2、如果要调…