如何基于请求取消std::async创建的指定ID工作线程

核心思路 #

• 为每个工作线程分配唯一可哈希的ID（避免使用std::thread::get_id()，其类型不适合作为容器键）
• 维护一个线程安全的集合，存储需要被取消的线程ID
• 工作线程在长耗时任务的关键中断点（如循环迭代、子任务完成后）主动检查自身ID是否在取消集合中，若存在则安全退出

完整代码实现 #

#include <iostream>
#include <thread>
#include <unordered_set>
#include <mutex>
#include <atomic>
#include <chrono>
#include <vector>
#include <memory>
#include <algorithm>
#include <functional>

// 自定义线程ID类型，保证唯一性与可哈希性
using ThreadId = uint64_t;

class TaskScheduler {
private:
    std::mutex mtx_;
    std::unordered_set<ThreadId> cancel_ids_;
    std::atomic<ThreadId> next_id_{1}; // 原子自增生成唯一ID
    std::vector<std::pair<ThreadId, std::shared_ptr<std::thread>>> threads_;

    // 工作线程核心逻辑：包含定期取消检查
    void worker_loop(ThreadId tid, const std::function<void()>& task_unit) {
        try {
            while (true) {
                // 检查取消标记：加锁访问线程安全集合
                {
                    std::lock_guard<std::mutex> lock(mtx_);
                    if (cancel_ids_.count(tid)) {
                        std::cout << "Thread " << tid << " received cancel signal, exiting\n";
                        cancel_ids_.erase(tid);
                        return;
                    }
                }

                // 执行单次任务单元（替换为实际业务逻辑）
                task_unit();

                // 模拟任务耗时（实际场景可移除）
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
        } catch (const std::exception& e) {
            std::lock_guard<std::mutex> lock(mtx_);
            cancel_ids_.erase(tid);
            std::cout << "Thread " << tid << " exited with exception: " << e.what() << "\n";
        } catch (...) {
            std::lock_guard<std::mutex> lock(mtx_);
            cancel_ids_.erase(tid);
            std::cout << "Thread " << tid << " exited with unknown exception\n";
        }
    }

public:
    // 创建工作线程，返回唯一ID
    ThreadId create_worker(const std::function<void()>& task_unit) {
        ThreadId tid = next_id_++;
        auto thread_ptr = std::make_shared<std::thread>(
            &TaskScheduler::worker_loop, this, tid, task_unit
        );

        std::lock_guard<std::mutex> lock(mtx_);
        threads_.emplace_back(tid, thread_ptr);
        std::cout << "Created worker thread: " << tid << "\n";
        return tid;
    }

    // 根据ID取消指定线程
    bool cancel_worker(ThreadId tid) {
        std::lock_guard<std::mutex> lock(mtx_);
        // 检查线程是否存在
        auto thread_iter = std::find_if(threads_.begin(), threads_.end(),
            [tid](const auto& entry) { return entry.first == tid; });
        
        if (thread_iter == threads_.end()) {
            std::cout << "Error: Thread " << tid << " does not exist\n";
            return false;
        }

        // 添加到取消集合
        cancel_ids_.insert(tid);
        std::cout << "Marked thread " << tid << " for cancellation\n";
        return true;
    }

    // 等待所有线程完成并清理资源
    void wait_all_workers() {
        std::lock_guard<std::mutex> lock(mtx_);
        for (auto& [tid, thread_ptr] : threads_) {
            if (thread_ptr->joinable()) {
                thread_ptr->join();
            }
        }
        threads_.clear();
    }

    ~TaskScheduler() {
        wait_all_workers();
    }
};

// 示例：业务任务单元
void sample_business_task() {
    std::cout << "Thread " << std::this_thread::get_id() << " executing task unit\n";
}

int main() {
    TaskScheduler scheduler;

    // 创建3个工作线程
    ThreadId tid1 = scheduler.create_worker(sample_business_task);
    ThreadId tid2 = scheduler.create_worker(sample_business_task);
    ThreadId tid3 = scheduler.create_worker(sample_business_task);

    // 运行3秒后取消线程2
    std::this_thread::sleep_for(std::chrono::seconds(3));
    scheduler.cancel_worker(tid2);

    // 再运行2秒后取消线程1
    std::this_thread::sleep_for(std::chrono::seconds(2));
    scheduler.cancel_worker(tid1);

    // 最后取消线程3
    std::this_thread::sleep_for(std::chrono::seconds(2));
    scheduler.cancel_worker(tid3);

    return 0;
}

关键说明 #

1. 唯一ID生成：使用std::atomic<ThreadId>自增生成ID，确保多线程下的唯一性，同时避免依赖std::thread::get_id()的不可哈希类型。
2. 线程安全集合：用std::mutex保护cancel_ids_，防止多线程并发修改导致的竞态条件。
3. 主动取消机制：工作线程在每次任务单元执行后检查取消标记，主动退出而非被强制终止——强制终止线程会引发资源泄漏、数据损坏等风险，主动退出是C++中安全终止线程的唯一推荐方式。
4. 资源清理：线程退出时自动从取消集合中移除ID，避免无效标记；析构函数中等待所有线程完成，防止程序退出时残留未终止的线程。

内容的提问来源于stack exchange，提问作者Bubblycentaur