CPP_Core/Core/Base/ThreadManager.cpp

#include "ThreadManager.h"
#include <chrono>
#include <iostream>

#include "../system/export.h"

using namespace std::chrono;


void Detach_Thread_Manager::start_thread(const std::string& name, const std::function<void(std::atomic<bool>&)>& f) {
    auto iter = thread_map.find(name);
    if (iter == thread_map.end())
    {
        auto t = new Detach_Thread(name, [f, name, this](std::atomic<bool>& running) {
            {
                std::ostringstream oss;
                oss << "\t" + name + "线程[id:" << std::this_thread::get_id() << "] 启动！\n";
                std::cout << oss.str() << std::flush;
            }

            f(running);
            {
                std::ostringstream oss;
                oss <<  " 【" + name + "】 线程 ";
                auto end = std::chrono::high_resolution_clock::now();
                auto duration = duration_cast<std::chrono::milliseconds>(end - exit_time);  // 转换为毫秒
                oss << "耗时: " + std::to_string(duration.count()) +  " 毫秒" + " 退出！ "
                << "thread id:[" << std::this_thread::get_id() << "] ";
                std::cout << oss.str() << std::endl;
            }
        });
        thread_map[name] = t;
        t->start();
    } else
    {
        std::cerr << "Detach_Thread_Manager append_thread error! Thread " << name << " already exists." << std::endl;
    }
}
void Detach_Thread_Manager::exit_thread(const std::string& name) {
    bool ok = false;
    for (auto it = thread_map.begin(); it != thread_map.end(); it++)
    {
        if (it->first == name)
        {
            auto t = it->second;
            t->running.store(false, std::memory_order_release);
            while (!t->exit.load(std::memory_order_acquire))
            {
                // 等待所有线程退出
            }
            std::cout << "\t" + name + "线程关闭！\n";
            delete it->second;
            thread_map.erase(it);
            ok = true;
            break;
        }
    }
    if (!ok)
    {
    }
}
Detach_Thread::Detach_Thread(const std::string &name, const std::function<void(std::atomic<bool> &running)> &func)  {
    this->name = name;
    thread = std::thread([func, this, name]() {
        Psc::set_current_thread_name(name);
        func(running);
        // 标志守护函数已经退出
        this->exit.store(true, std::memory_order_release);
    });
}
void Detach_Thread::start()  {
    thread.detach();
}
void Detach_Thread_Manager::test_and_start_thread(const std::string                                     &name,
const std::function<void(std::atomic<bool> &running)> &func)  {
    if (thread_map.find(name) == thread_map.end())
    {
        start_thread(name, func);
    }
}
void Detach_Thread_Manager::test_and_stop_thread(const std::string &name)  {
    if (thread_map.find(name) != thread_map.end())
    {
        exit_thread(name);
    }
}

void Detach_Thread_Manager::stop_all_thread(size_t timeout_milliseconds, std::set<std::string> excluded_thread) {
    auto start = std::chrono::steady_clock::now();
    std::ostringstream oss;
    oss << "开始等待所有子线程关闭  共有" << thread_map.size() << "个线程正在运行：  " << std::endl;
    std::vector<Detach_Thread*> close_list;
    std::set<Detach_Thread*> need_close_set;

    int index = 0;
    for (auto it = thread_map.begin(); it != thread_map.end(); it++) {
        if (excluded_thread.count(it->first) != 0) {
            oss << "index:[" << index << "] 排除:" << it->first << "  " << it->second->name << std::endl;
        } else {
            close_list.push_back(it->second);
            oss << "index:[" << index << "] 需要关闭：" << it->first << "  " << it->second->name << std::endl;
            need_close_set.insert(it->second);
        }
        index++;
    }

    // 设置退出时间
    auto exit_time = std::chrono::high_resolution_clock::now();

    for (auto it : close_list) {
        oss << it->name << "  ";
        it->running.store(false, std::memory_order_release);
    }
    oss << std::endl;
    std::cout << oss.str() << std::endl;

    while (true) {
        // 创建一个临时集合来存储已经关闭的线程
        std::set<Detach_Thread*> to_remove;

        for (auto it = need_close_set.begin(); it != need_close_set.end(); /* no increment here */) {
            if ((*it)->exit.load(std::memory_order_relaxed)) {
                // 如果线程已退出，则添加到移除列表
                to_remove.insert(*it);
                it = need_close_set.erase(it); // 在迭代器有效时删除元素
            } else {
                ++it; // 继续遍历
            }
        }

        // 如果没有需要关闭的线程，退出循环
        if (need_close_set.empty()) {
            std::cout << "需要关闭的子线程已全部关闭完成！" << std::endl;
            break;
        }

        // 如果还有线程未关闭，检查超时
        auto now = std::chrono::steady_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(now - start).count();
        long long remaining_time = timeout_milliseconds - duration;

        std::ostringstream nc_oss;
        for (auto need_close : need_close_set) {
            nc_oss << need_close->name << "  ";
        }

        if (remaining_time > 0) {
            if (fl.test()) {
                std::cout << "=================  剩余时间: " << remaining_time << " 毫秒" << nc_oss.str() << std::endl;
            }
        } else {
            std::cout << " ================= 超时！ 强制退出！" << std::endl;
            break;
        }

        // 等待一段时间后再次检查
        std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 每100毫秒检查一次
    }

    // 清理已关闭的线程
    for (auto it : close_list) {
        delete it;
    }

    thread_map.clear();
}

Detach_Thread* Detach_Thread_Manager::get_thread(const std::thread::id thread_id) {
    for (auto it = thread_map.begin(); it != thread_map.end(); it++) {
        if (it->second->thread.get_id() == thread_id) {
            return it->second;
        }
    }
    return nullptr;
}