恋恋风辰的官方博客

实现线程安全的查找表
Posted on 2023-11-26 11:00:02 by 恋恋风辰 | In C++ | 原文链接

简介

前文介绍了线程安全的队列和栈，本文继续介绍线程安全的查找结构，实现一个类似线程安全的map结构，但是map基于红黑树实现，假设我们要增加或者删除节点，设计思路是依次要删除或增加节点的父节点，然后修改子节点数据 。尽管这种思路可行，但是难度较大，红黑树节点的插入要修改多个节点的关系。另外加锁的流程也是锁父节点，再锁子节点，尽管在处理子节点时我们已经处理完父节点，可以对父节点解锁，继续对子节点加锁，这种情况锁的粒度也不是很精细，考虑用散列表实现。

散列表

散列表（Hash table，也叫哈希表），是根据键（Key）而直接访问在存储器存储位置的数据结构。 也就是说，它通过计算出一个键值的函数，将所需查询的数据映射到表中一个位置来让人访问，这加快了查找速度。 这个映射函数称做散列函数，存放记录的数组称做散列表。

举个例子：

假如我们一共有 50 人参加学校的数学竞赛，然后我们为每个学生分配一个编号，依次是 1 到 50.

如果我们想要快速知道编号对应学生的信息，我们就可以用一个数组来存放学生的信息，编号为 1 的放到数组下标为 1 的位置，编号为 2 的放到数组下标为 2 的位置，依次类推。

现在如果我们想知道编号为 20 的学生的信息，我们只需要把数组下标为 20 的元素取出来就可以了，时间复杂度为 O(1)，是不是效率非常高呢。

但是这些学生肯定来自不同的年级和班级，为了包含更详细的信息，我们在原来编号前边加上年级和班级的信息，比如 030211 ，03 表示年级，02 表示班级，11 原来的编号，这样我们该怎么存储学生的信息，才能够像原来一样使用下标快速查找学生的信息呢？

思路还是和原来一样，我们通过编号作为下标来储存，但是现在编号多出了年级和班级的信息怎么办呢，我们只需要截取编号的后两位作为数组下标来储存就可以了。

这个过程就是典型的散列思想。其中，参赛学生的编号我们称之为键(key)，我们用它来标识一个学生。然后我们通过一个方法（比如上边的截取编号最后两位数字）把编号转变为数组下标，这个方法叫做散列函数（哈希函数），通过散列函数得到的值叫做散列值（哈希值）。

我们自己在设计散列函数的函数时应该遵循什么规则呢？

  1. 得到的散列值是一个非负整数
  2. 两个相同的键，通过散列函数计算出的散列值也相同
  3. 两个不同的键，计算出的散列值不同

虽然我们在设计的时候要求满足以上三条要求，但对于第三点很难保证所有不同的建都被计算出不同的散列值。有可能不同的建会计算出相同的值，这叫做哈希冲突。最常见的一些解决哈希冲突的方式是开放寻址法和链表法，我们这里根据链表法，将散列函数得到相同的值的key放到同一个链表中。

如下图

https://cdn.llfc.club/1700962817978.jpg

当我们根据key值的后两位计算编号，将编号相同的放入一个链表，比如030211和030311是一个编号，所以将其放入一个链表。

同样的道理040213和060113是一个编号，放入一个链表。

设计思路

我们要实现上述逻辑，可以考虑将11，12，13等hash值放入一个vector中。多线程根据key计算得出hash值的过程并不需要加锁，可以实现并行计算。

但是对于链表的增删改查需要加锁。

所以我们考虑将链表封装为一个类bucket_type,支持数据的增删改查。

我们将整体的查找表封装为threadsafe_lookup_table类，实现散列规则和调度bucket_type类。

代码实现

我们先实现内部的bucket_type类. 为了threadsafe_lookup_table可以访问他，所以将threadsafe_lookup_table设置为其友元类。

  1. class bucket_type
  2. {
  3.     friend class threadsafe_lookup_table;
  4. }

我们需要用链表存储键值结构，所以我们可以在bucket_type中添加一个链表存储键值结构。

  1.     //存储元素的类型为pair，由key和value构成
  2.     typedef std::pair<Key, Value> bucket_value;
  3.     //由链表存储元素构
  4.     typedef std::list<bucket_value> bucket_data;
  5.     //链表的迭代器
  6.     typedef typename bucket_data::iterator bucket_iterator;
  7.     //链表数据
  8.     bucket_data data;
  9.     //改用共享锁
  10.     mutable std::shared_mutex mutex;

并且添加了互斥量用于控制链表的读写互斥操作，但是我们采用的是共享互斥量，可以实现读写锁，保证读的时候可以并发读。

接下来我们封装一个私有的查找接口，用来内部使用。

  1. //查找key值，找到返回对应的value，未找到则返回默认值
  2. bucket_iterator find_entry_for(const Key & key)
  3. {
  4.     return std::find_if(data.begin(), data.end(),
  5.                 [&](bucket_value const& item)
  6.                 {return item.first == key; });
  7. }

然后我们分别实现返回查找的值操作，以及添加操作，并且删除操作

  1.     //查找key值，找到返回对应的value，未找到则返回默认值
  2.     Value value_for(Key const& key, Value const& default_value) 
  3.     {
  4.         std::shared_lock<std::shared_mutex> lock(mutex);
  5.         bucket_iterator const found_entry = find_entry_for(key);
  6.         return (found_entry == data.end()) ?
  7.                 default_value : found_entry->second;
  8.     }
  9.     //添加key和value，找到则更新，没找到则添加
  10.     void add_or_update_mapping(Key const& key, Value const& value)
  11.     {
  12.         std::unique_lock<std::shared_mutex> lock(mutex);
  13.         bucket_iterator const found_entry = find_entry_for(key);
  14.         if (found_entry == data.end())
  15.         {
  16.             data.push_back(bucket_value(key, value));
  17.         }
  18.         else
  19.         {
  20.             found_entry->second = value;
  21.         }
  22.     }
  23.     //删除对应的key
  24.     void remove_mapping(Key const& key)
  25.     {
  26.         std::unique_lock<std::shared_mutex> lock(mutex);
  27.         bucket_iterator const found_entry = find_entry_for(key);
  28.         if (found_entry != data.end())
  29.         {
  30.             data.erase(found_entry);
  31.         }
  32.     }

这样我们设计完成了bucket_type类。

接下来我们设计threadsafe_lookup_table类。我们用一个vector存储上面的bucket_type类型。 因为我们要计算hash值，key可能是多种类型string, int等，所以我们采用std的hash算法作为散列函数即可.

  1. class threadsafe_lookup_table{
  2. private:
  3.     //用vector存储桶类型
  4.     std::vector<std::unique_ptr<bucket_type>> buckets;
  5.     //hash<Key> 哈希表 用来根据key生成哈希值
  6.     Hash hasher;
  8.     //根据key生成数字，并对桶的大小取余得到下标，根据下标返回对应的桶智能指针
  9.     bucket_type& get_bucket(Key const& key) const
  10.     {
  11.         std::size_t const bucket_index = hasher(key) % buckets.size();
  12.         return *buckets[bucket_index];
  13.     }
  14. };

get_bucket函数不需要加锁，各个线程可以并行计算哈希值，取出key对应的桶。如果多线程调用同一个bucket的增删改查，就通过bucket内部的互斥解决线程安全问题。
接下来我们完善threadsafe_lookup_table的对外接口

  1.     threadsafe_lookup_table(
  2.         unsigned num_buckets = 19, Hash const& hasher_ = Hash()) :
  3.         buckets(num_buckets), hasher(hasher_)
  4.     {
  5.         for (unsigned i = 0; i < num_buckets; ++i)
  6.         {
  7.             buckets[i].reset(new bucket_type);
  8.         }
  9.     }
  11.     threadsafe_lookup_table(threadsafe_lookup_table const& other) = delete;
  12.     threadsafe_lookup_table& operator=(
  13.         threadsafe_lookup_table const& other) = delete;
  15.     Value value_for(Key const& key,
  16.         Value const& default_value = Value()) 
  17.     {
  18.         return get_bucket(key).value_for(key, default_value);
  19.     }
  21.     void add_or_update_mapping(Key const& key, Value const& value)
  22.     {
  23.         get_bucket(key).add_or_update_mapping(key, value);
  24.     }
  26.     void remove_mapping(Key const& key)
  27.     {
  28.         get_bucket(key).remove_mapping(key);
  29.     }

除此之外我们可将当前查找表的副本作为一个map返回

  1. std::map<Key, Value> get_map() 
  2. {
  3.     std::vector<std::unique_lock<std::shared_mutex>> locks;
  4.     for (unsigned i = 0; i < buckets.size(); ++i)
  5.     {
  6.         locks.push_back(
  7.                 std::unique_lock<std::shared_mutex>(buckets[i]->mutex));
  8.     }
  9.     std::map<Key, Value> res;
  10.     for (unsigned i = 0; i < buckets.size(); ++i)
  11.     {
  12.         //需用typename告诉编译器bucket_type::bucket_iterator是一个类型，以后再实例化
  13.         //当然此处可简写成auto it = buckets[i]->data.begin();
  14.         typename bucket_type::bucket_iterator it = buckets[i]->data.begin();
  15.         for (;it != buckets[i]->data.end();++it)
  16.         {
  17.             res.insert(*it);
  18.         }
  19.     }
  20.     return res;
  21. }

测试与分析

我们自定义一个类

  1. class MyClass
  2. {
  3. public:
  4.     MyClass(int i):_data(i){}
  6.     friend std::ostream& operator << (std::ostream& os, const MyClass& mc){
  7.         os << mc._data;
  8.         return os;
  9.     }
  12. private:
  13.     int _data;
  14. };

接下来我们实现一个函数做测试

  1. void TestThreadSafeHash() {
  2.     std::set<int> removeSet;
  3.     threadsafe_lookup_table<int, std::shared_ptr<MyClass>> table;
  4.     std::thread t1([&]() {
  5.         for(int i = 0; i < 100; i++)
  6.         {
  7.            auto class_ptr =  std::make_shared<MyClass>(i); 
  8.             table.add_or_update_mapping(i, class_ptr);
  9.         }
  10.     });
  12.     std::thread t2([&]() {
  13.         for (int i = 0; i < 100; )
  14.         {
  15.             auto find_res = table.value_for(i, nullptr);
  16.             if(find_res)
  17.             {
  18.                 table.remove_mapping(i);
  19.                 removeSet.insert(i);
  20.                 i++;
  21.             }
  23.             std::this_thread::sleep_for(std::chrono::milliseconds(10));
  24.         }
  25.     });
  27.     std::thread t3([&]() {
  28.         for (int i = 100; i < 200; i++)
  29.         {
  30.             auto class_ptr = std::make_shared<MyClass>(i);
  31.             table.add_or_update_mapping(i, class_ptr);
  32.         }
  33.         });
  36.     t1.join();
  37.     t2.join();
  38.     t3.join();
  40.     for(auto & i : removeSet)
  41.     {
  42.         std::cout << "remove data is " << i << std::endl;
  43.     }
  45.    auto copy_map =  table.get_map();
  46.     for(auto & i : copy_map)
  47.     {
  48.         std::cout << "copy data is " << *(i.second) << std::endl;
  49.     }
  50. }

t1用来向map中添加数据(从0到99)，t2用来从map中移除数据(从0到99)，如果map中未找到则等待10ms继续尝试，t3则继续继续添加数据(从100到199).
然后分别打印插入的集合和获取的map中的数值。
打印可以看到输出插入集合为(0~99)，copy的map集合为(100~199).

我们分析一下上述查找表的优劣

1 首先我们的查找表可以支持并发读，并发写，并发读的时候不会阻塞其他线程。但是并发写的时候会卡住其他线程。基本的并发读写没有问题。
2 但是对于bucket_type中链表的操作加锁精度并不精细，因为我们采用的是std提供的list容器，所以增删改查等操作都要加同一把锁，导致锁过于粗糙。

下一节会介绍支持并发读写的自定义链表，可以解决bucket_type中的list锁精度不够的短板。

总结

本文介绍了线程安全情况下栈和队列的实现方式

源码链接:

https://gitee.com/secondtonone1/boostasio-learn/tree/master/concurrent/day15-threadsafehash

视频链接

https://space.bilibili.com/271469206/channel/collectiondetail?sid=1623290

热门评论
点击查看更多评论 没有更多评论了,点击收起 展开评论

热门文章

  1. vscode搭建windows C++开发环境
    喜欢(596) 浏览(82027)

  2. 聊天项目(28) 分布式服务通知好友申请
    喜欢(507) 浏览(6016)

  3. 使用hexo搭建个人博客
    喜欢(533) 浏览(11715)

  4. Linux环境搭建和编码
    喜欢(594) 浏览(13348)

  5. Qt环境搭建
    喜欢(517) 浏览(24312)

最新评论

  1. 聊天项目(15) 客户端实现TCP管理者 lkx:已经在&QTcpSocket::readyRead 回调函数中做了处理了的。
  2. 解决博客回复区被脚本注入的问题 secondtonone1:走到现在我忽然明白一个道理，无论工作也好生活也罢，最重要的是开心，即使一份安稳的工作不能给我带来事业上的积累也要合理的舍弃，所以我还是想去做喜欢的方向。
  3. interface应用 secondtonone1:interface是万能类型，但是使用时要转换为实际类型来使用。interface丰富了go的多态特性，也降低了传统面向对象语言的耦合性。
  4. string类 WangQi888888:确实错了，应该是！isspace(sind[index]). 否则不进入循环，还是原来的字符串“some string”
  5. 创建项目和编译 secondtonone1:谢谢支持
  6. 构造函数 secondtonone1:构造函数是类的基础知识，要着重掌握
  7. 聊天项目(9) redis服务搭建 pro_lin:redis线程池的析构函数，除了pop出队列，还要free掉redis连接把
  8. 可变参数模板+异步队列实现异步打印功能 熊二:学到了%%%
  9. C++ 并发三剑客future, promise和async Yunfei:大佬您好，如果这个线程池中加入的异步任务的形参如果有右值引用，这个commit中的返回类型推导和bind绑定就会出现问题，请问实际工程中，是不是不会用到这种任务，如果用到了，应该怎么解决？
  10. Qt 对话框 Spade2077:QDialog w(); //这里是不是不需要带括号
  11. visual studio配置boost库 一giao里我离giaogiao:请问是修改成这样吗：.\b2.exe toolset=MinGW
  12. 类和对象 陈宇航:支持！！！！
  13. protobuf配置和使用 熊二:你可以把dll放到系统目录，也可以配置环境变量，还能把dll丢到lib里
  14. boost::asio之socket的创建和连接 项空月:发现一些错别字 :每隔vector存储  是不是是每个. asio::mutable_buffers_1 o或者    是不是多打了个o
  15. 堆排序 secondtonone1:堆排序非常实用，定时器就是这个原理制作的。
  16. 答疑汇总(thread,async源码分析) Yagus:如果引用计数为0，则会执行 future 的析构进而等待任务执行完成，那么看到的输出将是 这边应该不对吧，std::future析构只在这三种情况都满足的时候才回block: 1.共享状态是std::async 创造的(类型是_Task_async_state) 2.共享状态没有ready 3.这个future是共享状态的最后一个引用 这边共享状态类型是“_Package_state”，引用计数即使为0也不应该block啊
  17. 再谈单例模式 secondtonone1:是的，C++11以后返回局部static变量对象能保证线程安全了。
  18. 面试题汇总(一) secondtonone1:看到网络上经常提问的go的问题，做了一下汇总，结合自己的经验给出的答案，如有纰漏，望指正批评。
  19. 利用栅栏实现同步 Dzher:作者你好！我觉得 std::thread a(write_x); std::thread b(write_y); std::thread c(read_x_then_y); std::thread d(read_y_then_x); 这个例子中的assert fail并不会发生，原子变量设定了非relaxed内存序后一个线程的原子变量被写入，那么之后的读取一定会被同步的，c和d线程中只可能同时发生一个z++未执行的情况，最终z不是1就是2了，我测试了很多次都没有assert，请问我这个观点有什么错误，谢谢！
  20. 无锁并发队列 TenThousandOne:_head  和 _tail  替换为原子变量。那里pop的逻辑，val = _data[h] 可以移到循环外面吗
  21. 网络编程学习方法和图书推荐 Corleone:啥程度可以找工作
  22. 处理网络粘包问题 zyouth: //消息的长度小于头部规定的长度，说明数据未收全，则先将部分消息放到接收节点里 if (bytes_transferred < data_len) { memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred); _recv_msg_node->_cur_len += bytes_transferred; ::memset(_data, 0, MAX_LENGTH); _socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH), std::bind(&CSession::HandleRead, this, std::placeholders::_1, std::placeholders::_2, shared_self)); //头部处理完成 _b_head_parse = true; return; } 把_b_head_parse = true;放在_socket.async_read_some前面是不是更好
  23. Qt MVC结构之QItemDelegate介绍 胡歌-此生不换:gpt, google
  24. 聊天项目(7) visualstudio配置grpc diablorrr:cmake文件得改一下 find_package(Boost REQUIRED COMPONENTS system filesystem)，要加上filesystem。在target_link_libraries中也同样加上

其他链接

  1. GitHub
  2. 个人博客
  3. 博客园
  4. CSDN

个人公众号

个人微信