并发消息处理按时间顺序排列

Question

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

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

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

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

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

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

感谢任何cmets！

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++;
    }
}

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