世界上最简单的无锁Hash Table

最新推荐文章于 2023-03-10 13:15:46 发布

zhangyifei216

最新推荐文章于 2023-03-10 13:15:46 发布

阅读量2.1k

点赞数 1

分类专栏：高性能并发编程文章标签： hashtable 无锁 lock-free

本文链接：https://blog.csdn.net/zhangyifei216/article/details/52862466

版权

高性能并发编程专栏收录该内容

3 篇文章

订阅专栏

本文介绍了如何使用C++11实现一个限制性的无锁HashMap，仅支持32位整型key和value，且固定大小。通过开放地址法和线性探测解决冲突，利用原子操作和CAS保证线程安全。详细阐述了插入和查找操作的实现，并提供了完整的代码实现链接。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

一个无锁的hash table是把双刃剑，它可以为某些应用程序提供性能上的大幅提升，它的缺点就是过于复杂了。Dr. Cliff Click使用Java写了第一个可工作的无锁hash table，并在2007年的时候公布其源码，并且在同一年在Google发表了演讲。

原理分析

本文将通过使用C++11来编写一个无锁的hash table，作为一个程序员我们通常会贫直觉去写一些通用的数据结构并尽可能的去重用代码。这不是一件坏事，但是如果我们把它变成一个必须要做的一件事，那么它可能会阻碍我们。本文将会去实现一个极端的，有限的，用于一些特殊场合的无锁hash table，它有一些设计上的限制。

仅支持32位的整型key，映射到32位的整型value
所有的key必须是非零的
所有的value必须是非零的
hash table的大小事固定的，并且必须是指数倍
仅仅允许SetItem和GetItem
没有删除操作

不过请放心，一旦你掌握了这种受限的Hash table，就可以逐步删除上面每一个限制，而无需从根本上改变方法。实现的原理都是相通的。

有许多方法可以用来实现一个hash table，一种简单的实现就是利用数组存放每一个条目，每一个条目包含一个key/value键值对如下:

struct Entry
{
    std::atomic<uint32_t> key;
    std::atomic<uint32_t> value;
};

使用这种方法实现hash table的时候，当key发生冲突时需要使用开放地址法来解决冲突，所有的元素最终都放在数组中。下面是我本文所采用的MurmurHash3 hash算法，对于输入是整数的场景下它许相当的快。

inline static uint32_t integerHash(uint32_t h) {
  h ^= h >> 16;
  h *= 0x85ebca6b;
  h ^- h >> 13;
  h *= 0xc2b2ae35;
  h ^= h >> 16;
  return h;
}

对于相同的key来说，无论是插入还是查找，开始探测的下标都是一样的，和普通数组不同的是，hash table每次都是先对key进行hash，定位到起始下标，然后开始进行线性探测，而普通数组每次都是从0开始。而对于不同的key，开始探测的下标都是不同的。最终的结果就是所有的条目分布在数组的各个位置。多线程可以安全的并行的插入和查找。

使用开放地址法进行线性探测，我认为这是一种对于实现无锁算法来友好的hash table技术，事实上这也是

Dr. Cliff Click 基于Java实现的无锁hash table所使用的技术。

代码实现

下面是SetItem的具体实现，也就是往hash table中插入元素，通过扫描数组中的每一个Entry的key是否是0，或者是匹配的key，来决定插入的位置。

void HashTable::SetItem(uint32_t key, uint32_t value) {
  uint32_t zero = 0;
  for (uint32_t idx = integerHash(key);; ++idx) {
    idx &= arraySize_ - 1;
    uint32_t prevKey = entry_[idx].key.load(std::memory_order_relaxed);
    if (prevKey != key) {
      if (prevKey != 0) {
        continue;
      }
      bool succ = entry_[idx].key.compare_exchange_strong(zero, key, std::memory_order_relaxed);
      if (succ)
          goto out;
      continue;
    }
out:    
    entry_[idx].value.store(value, std::memory_order_relaxed);
    return;
  }
}

上面的代码首先通过idx &= arraySize_ -1得到了这个key在数组中的起始下标，这里巧妙了利用了 &操作来取模，能使用这种方式的原因在于arraySize_是个2的指数倍，这是一种比较快的的取模方法，更详细的内容参见performance-pattern-modulo-and-powers-of-two。

紧接着对key进行了检查，不满足要求的就continue，如果和插入的key相同，则直接原子更新value即可，如果是0则进行CAS，原子的更新key，如果更新失败(其他线程并发插入导致更新失败)，就继续寻找下一个空闲的位置，否则就更新value值然后返回。下面是GetItem的实现，相比较SetItem要简单多了。

uint32_t HashTable::GetItem(uint32_t key) {
  for(uint32_t idx = integerHash(key);; idx++) {
    idx &= arraySize_ - 1;
    uint32_t prevKey = entry_[idx].key.load(std::memory_order_relaxed);
    if (prevKey == key)
      return entry_[idx].value.load(std::memory_order_relaxed);

    if (prevKey == 0)
      return 0;
  }
  return 0;
}

上面的代码就是找到匹配的Entry，然后获取其value返回即可，需要注意的就是if (prevKey == 0)这段代码，当查找的key是0的时候是return 0，而不是continue，原因在于本文使用的是线性探测解决Hash冲突，并且没有删除操作，所以在探测的路径上是不会有空闲的位置，也就是key为0的位置。

到此为止一个线程安全的，无锁的Hash Table就实现了，通过原子操作和CAS实现了无锁，完整代码见github