c++ - 按时间顺序排列的并发消息处理

Question

我想在性能方面优化用 C++ 编写的消息解码器。解码器完全按顺序设计。实际并行化的概念很简单:

一旦新数据到达某个套接字，就告诉线程池运行另一个线程来解码接收到的消息。

在每个线程结束时，将调用一个方法(即发出一个Qt信号)并传递一个在处理过程中创建的对象。

我的问题是:处理消息的长度和复杂性各不相同，因此线程完成的顺序可能与接收消息的顺序不同。换句话说，我需要在不使用线程安全容器的情况下就地序列化。

如何确保线程在完成后立即按正确的时间顺序调用上述方法，而不是将它们排入线程安全容器中？

我的第一个想法是创建与线程池中的线程一样多的互斥体，然后使用每个互斥体将“完成”信号从旧线程发送到新线程。

感谢任何意见!

Answer 1

如果您真的不想使用像 priority_queue 或一系列预保留缓冲区这样的数据结构并改为阻塞您的线程，您可以执行以下操作:

1. 将您的消息与指示其原始的索引配对位置并将其传递给线程池。
2. 使用一个通用的(例如全局的、原子的)计数器变量来指示最后处理的消息。
3. 让每个线程等待，直到该变量指示上一条消息已被处理。
4. 传递生成的对象并增加计数器

代码看起来像这样:

struct MsgIndexed {
    size_t idx;
    Msg msg;
};

//Single thread that receives all messages sequentially
void threadReceive() {
    for (size_t i = 1; true ; i++)
    {
        Msg m = readMsg();
        dipatchMsg(MsgIndexed{i,m});
    }
}

std::atomic<size_t> cnt=0;
//multiple worker threads that work in parallel
void threadWork() {
    while (1) {
        MsgIndexed msg = waitforMsg();
        Obj obj = processMsg(msg.msg);

        //Just for demonstration purposes. 
        //You probably don't want to use a spinlock here, but e.g. a condition variable instead     
        while (cnt != (msg.idx - 1u)) { std::this_thread::yield(); } 

        forwardObj(obj);
        cnt++;
    }
}

请注意，这是一个非常低效的解决方案，因为您的工作线程在完成实际工作后仍然需要等待。