c++ - 为线程安全写入映射中的队列时互斥

Question

我有一个 map<int, queue<int>>用一个线程写入其中，即将消息插入队列。他们的关键是指一个client_id ，队列保存客户端的消息。我希望使这个读写线程安全。

目前，写入它的线程会做这样的事情

map<int, queue<int>> msg_map;
if (msg_map.find(client_id) != msg_map.end())
{
    queue<int> dummy_queue;
    dummy_queue.push(msg); //msg is an int
    msg_map.insert(make_pair(client_id, dummy_queue);
}
else
{
    msg_map[client_id].push(msg);
}

有许多客户正在阅读和删除这张 map 。

if (msg_map.find(client_id) != msg_map.end())
{
    if (!msg_map.find(client_id)->second.empty())
    {
        int msg_rxed = msg_map[client_id].front();

        //processing message

        msg_map[client_id].pop();
    }
}

我正在阅读 this在互斥体上(之前没有使用过)我想知道何时何地我应该锁定互斥体。我的困惑在于他们正在访问单独的队列(保存在同一张 map 中)。我是锁定队列还是锁定 map ？

是否有标准/公认的方法来执行此操作 - 使用互斥锁是执行此操作的最佳方法吗？客户端线程为“0”，只有 1 个写入线程。

Answer 1

简化和优化您的代码

现在我们不关心互斥锁，我们稍后会在代码稍微清理一下时处理它(那时会更容易)。

首先，从您显示的代码来看，似乎没有理由使用有序的 std::map (对数复杂度)，你可以使用效率更高的 std::unordered_map (平均恒定时间复杂度)。选择完全取决于您，如果您不需要订购容器，您只需更改其声明:

std::map<int, std::queue<int>> msg_map;
// or
std::unordered_map<int, std::queue<int>> msg_map; // C++11 only though

现在， map 在设计上非常高效，但如果您坚持为每个操作进行查找，那么您将失去 map 的所有优势。

关于编写器线程，所有您的代码块(对于编写器)都可以通过以下行有效地替换:

msg_map[client_id].push(msg);

请注意 std::map 和 std::unordered_map 的 operator[] 定义为:

Inserts a new element to the container using key as the key and a default constructed mapped value and returns a reference to the newly constructed mapped value. If an element with key key already exists, no insertion is performed and a reference to its mapped value is returned.

关于您的读者线程，您不能直接使用 operator[]，因为如果当前不存在特定 client_id 的条目，它会创建一个新条目，因此，您需要缓存 find 返回的迭代器以便重用它，从而避免无用的查找:

auto iter = msg_map.find(client_id);
// iter will be either std::map<int, std::queue<int>>::iterator
//                  or std::unordered_map<int, std::queue<int>>::iterator
if (iter != msg_map.end()) {
    std::queue<int>& q = iter->second;
    if (!q.empty()) {
        int msg = q.front();
        q.pop();
        // process msg
    }
}

我之所以在处理消息之前立即弹出消息，是因为它会在我们添加互斥量时提高并发性(我们可以更快地解锁互斥量，这总是好的)。

使代码线程安全

@hmjd 关于多个锁(一个用于映射，一个用于每个队列)的想法很有趣，但是根据您向我们展示的代码，我不同意:您从额外的并发性中获得的任何好处很可能会被锁定队列互斥锁所需的额外时间(实际上，锁定互斥锁是一项非常昂贵的操作)，更不用说您必须处理的额外代码复杂性了。我敢打赌单个互斥锁(同时保护 map 和所有队列)效率更高。

顺便说一句，如果您想使用更高效的 std::unordered_map(std::map 不会受到影响，单个互斥体可以解决迭代器失效问题问题)。

假设使用 C++11，只需声明一个 std::mutex连同你的 map :

std::mutex msg_map_mutex;
std::map<int, std::queue<int>> msg_map; // or std::unordered_map

保护写入线程非常简单，只需在访问映射之前锁定互斥量即可:

std::lock_guard<std::mutex> lock(msg_map_mutex);
// the lock is held while the lock_guard object stays in scope
msg_map[client_id].push(msg);

保护读者线程几乎没有任何困难，唯一的技巧是您可能希望尽快解锁互斥锁以提高并发性，因此您将不得不使用 std::unique_lock (可以提前解锁)而不是 std::lock_guard (只有在超出范围时才能解锁):

std::unique_lock<std::mutex> lock(msg_map_mutex);
auto iter = msg_map.find(client_id);
if (iter != msg_map.end()) {
    std::queue<int>& q = iter->second;
    if (!q.empty()) {
        int msg = q.front();
        q.pop();
        // assuming you don't need to access the map from now on, let's unlock
        lock.unlock();
        // process msg, other threads can access the map concurrently
    }
}

如果您不能使用 C++11，则必须替换 std::mutex 等。使用您的平台提供的任何内容(pthreads、Win32 等)或使用 boost等效(与特定于平台的原语不同，它具有与新的 C++11 类一样可移植和易于使用的优点)。