c++ 并发线程中的类对象

Question

我试图启动一个有正在运行的线程的类。当我创建对象时，它会创建一组错误的对象。我真的不明白发生了什么。这个想法是用它们的线程创建 10 个对象并从线程 f2 获取消息。我需要一个按规格工作线程的队列。我使用互斥锁是因为我认为 c++ std:queue 不是线程安全的。问题是当 main 运行时它显示大多数对象都是相同的。我贴出代码：

class classOne {
public:
    classOne(int n, std::shared_ptr<std::mutex> prompr_mtx_ptr);
    ~classOne();

    int m_id;
    bool m_stop;
    std::shared_ptr<std::mutex> m_mtx;
    std::shared_ptr<std::thread> m_classOne_thread;
    std::shared_ptr<std::mutex> m_prompr_mtx_ptr;
    std::shared_ptr<std::queue<std::string>> m_queue;

    void classOne_thread_body();
};

classOne::~classOne() {

}

classOne::classOne(int n, std::shared_ptr<std::mutex> prompr_mtx_ptr):
        m_id(n),
        m_prompr_mtx_ptr(prompr_mtx_ptr)
{
    m_queue = std::make_shared<std::queue<std::string>>();
    m_mtx = std::make_shared<std::mutex>();
    m_stop = false;
    m_classOne_thread = std::make_shared<std::thread>(&classOne::classOne_thread_body, this);
}

void classOne::classOne_thread_body() {
    int watchDog = 0, debugDog = 0;
    m_prompr_mtx_ptr->lock();
    std::cout<<"[ Thread "<<std::to_string(m_id)<<" ] Started"<<std::endl;
    m_prompr_mtx_ptr->unlock();
    while (!m_stop) {
        m_mtx->lock();
        if (!m_queue->empty()) {
            m_prompr_mtx_ptr->lock();
            std::cout<<"[ Thread "<<std::to_string(m_id)<<" ] Received message: "<<m_queue->front()<<std::endl;
            m_prompr_mtx_ptr->unlock();
            m_queue->pop();
            watchDog = 0;
        }
        m_mtx->unlock();

        ++watchDog;
        if (watchDog>= 1000) { m_stop = true; }
        if (watchDog - debugDog > 100) {
            m_prompr_mtx_ptr->lock();
            std::cout<<"WatchDog: "<<watchDog<<std::endl;
            m_prompr_mtx_ptr->unlock();
            debugDog = watchDog;
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void f2(int n, std::vector<classOne> elements)
{
    for (int i = 0; i < 100; ++i) {
        for(std::vector<classOne>::iterator it=elements.begin(); it!=elements.end(); ++it) {
            it->m_mtx->lock();
            it->m_queue->push(std::string("Message "+std::to_string(i)+" to thread "+std::to_string(it->m_id)+" from "+std::to_string(n)));
            it->m_mtx->unlock();
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    }
}

int main()
{
    std::thread t1; // t1 is not a thread
    int generators = 10;
    std::vector<classOne> genClassOnes;
    std::shared_ptr<std::mutex> prompr_mtx_ptr = std::make_shared<std::mutex>();

    for (int i = 0; i < generators; i++) {
        genClassOnes.push_back(classOne(i, prompr_mtx_ptr));
    }

    std::thread t2(f2, 1, genClassOnes); // pass by value
    t2.join();

    for (std::vector<classOne>::iterator iter=genClassOnes.begin(); iter!=genClassOnes.end(); ++iter) {
        iter->m_classOne_thread->join();
    }
}

这是错误的输出：

[ Thread 3 ] Started
[ Thread 8 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Started
[ Thread 9 ] Received message: Message 0 to thread 9 from 1

Answer 1

std::vector<classOne> 要求 classOne 在您调用 push_back 或 resize 时是可复制的。而std::list<classOne> 不需要classOne 是可复制的，因此您可以删除所有这些shared_ptrs。

使用条件变量让线程知道队列中有可用的新元素。

工作示例：

#include <list>
#include <queue>
#include <mutex>
#include <string>
#include <thread>
#include <iostream>
#include <condition_variable>

class classOne {
    int m_id;
    std::mutex m_mtx;
    std::condition_variable m_cnd;
    std::mutex* const m_prompr_mtx_ptr;
    std::queue<std::string> m_queue;
    std::thread m_classOne_thread;

    void classOne_thread_body();

    template<class F>
    void log(F&& f) {
        std::lock_guard<std::mutex> lock(*m_prompr_mtx_ptr);
        f(std::cout);
    }

public:
    classOne(int n, std::mutex& prompr_mtx_ptr);

    ~classOne() {
        this->stop();
        m_classOne_thread.join();
    }

    void post(std::string msg) {
        {
            std::lock_guard<std::mutex> lock(m_mtx);
            m_queue.push(move(msg));
        }
        m_cnd.notify_one();
    }

    void stop() { this->post({}); }

    int get_id() const { return m_id; }
};

classOne::classOne(int n, std::mutex& prompr_mtx_ptr):
    m_id(n),
    m_prompr_mtx_ptr(&prompr_mtx_ptr),
    m_classOne_thread(&classOne::classOne_thread_body, this)
{}

void classOne::classOne_thread_body() {
    log([&](std::ostream& s) { s <<"[ Thread "<<std::to_string(m_id)<<" ] Started"<<std::endl; });
    for(std::string msg;;) {
        {
            std::unique_lock<std::mutex> lock(m_mtx);
            while(m_queue.empty())
                m_cnd.wait(lock);
            msg = m_queue.front();
            m_queue.pop();
        }
        if(msg.empty())
            break;
        log([&](std::ostream& s) { s <<"[ Thread "<<std::to_string(m_id)<<" ] Received message: "<< msg << std::endl; });
    }
}

void f2(int n, std::list<classOne>& elements) {
    for (int i = 0; i < 100; ++i) {
        for(auto& elem : elements)
            elem.post("Message "+std::to_string(i)+" to thread "+std::to_string(elem.get_id())+" from "+std::to_string(n));
    }
}

int main() {
    int generators = 10;
    std::mutex prompr_mtx_ptr;

    std::list<classOne> genClassOnes;
    for(int i = 0; i < generators; i++)
        genClassOnes.emplace_back(i, prompr_mtx_ptr);

    std::thread t2(f2, 1, std::ref(genClassOnes)); // pass by reference
    t2.join();
}