【C++杂货铺】再谈哈希算法：位图 | 布隆过滤器 | 哈希切分

一、位图

1.1 一道面试题

给40亿个不重复的无符号整数，没排过序。给一个无符号整数，如何快速判断一个数是否在这40亿个数中。

解决方案：

• 遍历：遍历这40亿个整数，去判断该数是否在这40亿个整数中。时间复杂度是 O ( N ) O(N)O(N)。
  
• set：用 set 将这40亿个整数存起来，然后去判断这个数是否在 set 中。时间复杂度是 O ( l o g 2 N ) O(log2^N)O(log2N)。
  
• 排序 + 二分查找：快排的时间复杂度是 O ( N ∗ l o g 2 N ) O(N*log2^N)O(N∗log2N)，二分查找的时间复杂度是 O ( l o g 2 N ) O(log2^N)O(log2N)。
  

存在的问题：40亿个整数是160亿Byte。由 1 G = 1024 M B = 1024 ∗ 1024 K b = 1024 ∗ 1024 ∗ 1024 B y t e 1G = 1024MB = 1024*1024Kb = 1024*1024*1024 Byte1G=1024MB=1024∗1024Kb=1024∗1024∗1024Byte，推出 10 亿 B y t e ≈ 1 G 10亿Byte≈1G10亿Byte≈1G。所以 160 亿 B y t e ≈ 16 G 160亿Byte≈16G160亿Byte≈16G（不到 16G）。上面这三种方法都需要将这40亿个整数先存储起来，这就需要从内存中分配 16G 空间来存储（假设单纯的用数组来存储，如果使用 set，set 中的节点不止存储整型数据，还会存储一些附带的东西，比如指针，最终需要的内存空间是远大于 16G 的），但是普通电脑的内存大小就是 8G 或者 16G，上面这三种做法显然是不符合实际的。那该怎么去解决呢？这道题目本质上是让我们去判断在不在，因此我们无需将这 40亿 个整数都存起来，只需要用一个标志位去判断在不在，比如 1 表示在，0 表示不在。存储标志位的最小单位是 bit。大体思路有了，下面引出位图的概念，这道题目需要使用位图去解决。

1.2 位图的概念

所谓位图，这里位指的就是比特位，位图就是用每一个比特位来存放某种状态，适用于海量数据，数据无重复的场景。通常是用来判断某个数据存不存在的。

如上图所示，要判断一个数字是否在 array 数组中，这里我们可以借助位图来实现。这里我们采用直接定址法，即 array 中的每一个数字都对应一个比特位，互相不重复。这就要求位图的大小是 array 数组中整型变量的范围。array 中整型的范围是 [1，22]，里面一共包含 22 个整型，因此这里的位图需要 22 个比特位。一个字符型 char 会向内存申请一个字节（8比特），所以这里我们创建一个可以存储 3 个字符的字符数组来充当位图。有了位图之后，接下来我们就需要对位图中的每一位进行标记，将 array 中的整型所对应的位图中的位设置成 1。然后给我们一个整型数据 x，要判断 x 是否在 array 中，我们只需要去判断 x 对应位图中的那个位上是 1 还是 0，如果是 1 就说明 array 中存在 x 这个整型，如果是 0 说明 array 中不存在 x 这个整型。

回到上面的面试题：先创建一个位图，位图的大小取决于数据的范围，面试题中说的是40亿个不重复的无符号整型，所以这里的数据范围就是无符号整型的范围，即 0~4294967295。因此我们需要向内存申请 4294967295 个比特位也就是 2 32 2^{32}232 个比特位。2 3 2^323 个比特位就是一个字节，所以 2 32 2^{32}232 个比特位就是 2 29 2^{29}229 个字节。由 1 G = 1024 M B = 1024 ∗ 1024 K b = 1024 ∗ 1024 ∗ 1024 B y t e = 2 30 B y t e 1G = 1024MB = 1024*1024Kb = 1024*1024*1024 Byte = 2^{30} Byte1G=1024MB=1024∗1024Kb=1024∗1024∗1024Byte=230Byte，推出 2 29 B y t e = 0.5 G = 500 M B 2^{29}Byte = 0.5G = 500MB229Byte=0.5G=500MB。对普通电脑来说，向内存申请 0.5G 的空间还是轻轻松松的。

1.3 位图的模拟实现

template<size_t N>//这里的 N 是一个非类型模板参数，表示要开 N 个比特位
class bitset
{
public:
  bitset()
  {
    size_t size = N / 32 + 1;
    _a.resize(size, 0);
    //最坏情况下，N / 32 能够整除，那就会浪费一个整型的大小，即四个字节。
  }
  //把 x 映射的那个比特位标记成1
  void set(size_t x)
  {
    size_t i = x / 32;
    size_t j = x % 32;
    _a[i] |= (1 << j);//位运算的左移和右移不是方向，左移是往高位进行移动，右移是往地位进行移动，至于底层到底是大端机还是小端机我们根本不用关心是怎么移动的
  }
  //把 x 映射的那个比特位标记成0
  void reset(size_t x)
  {
    size_t i = x / 32;
    size_t j = x % 32;
    _a[i] &= ~(1 << j);
  }
  //检测 x 映射的那个比特位是 0 还是 1
  bool test(size_t x)
  {
    size_t i = x / 32;
    size_t j = x % 32;
    return _a[i] & (1 << j);
  }
private:
  vector<int> _a;
};

小Tips：C++ 中没有一个类型的大小是一个比特位，因此我们采用一个整型数组的方式，一个整型 4 个字节，32 个比特位。bitset 这个类采用了非类型模板参数，用户将需要开辟多少个比特位传进来，然后开辟对应大小的数组。其次位图中的一个重点内容就是找到 x 对应的比特位，这里分两步，第一步要先找到 x 属于数组中的哪一个整型，用 x / 32 x / 32x/32 即可以得到，然后需要确定 x 对应该整型的哪一位，用 x xx % 32 3232 即可以得到。 找到对应的比特位后，还要对该比特位进行置 0 或置 1 操作来进行标记，因为我们的底层是用的整型数组，所以对某一个比特位进行置 0 或置 1 操作，本质上是对整型数组中的某一个整型进行位操作。

1.4 位图的应用

1.4.1 给定100亿个整数，设计算法找到只出现一次的整数

思路：回顾一下上面 40亿 个整数判断在不在的问题，我们的做法是让所有的整数都对应一个比特位，这个比特位用来标记该整型是否出现在这 40亿 个整型中。这里为什么每一个整型只需要对应一个比特位？因为一个整型只有在这 40亿 个数里面和不在这 40亿 个数里面两种情况，这两种情况我们分别可以用 1 和 0 来表示，而 1 和 0 都只占用一个比特位。

回到这道题目，我们任然可以采用标记位的方式来处理。对于一个整数来说，它在这 100亿 个整数中出现的情况有三种：出现了 0 次，出现了 1 次，出现了多次（两次及以上）。这道题目我们只需要这三种标记即可，一个比特位可以表示两个标记，那要想能够表示三个标记就至少需要两个比特位了，两个比特位一共有四种组合：00、01、10、11，完全够用了。总结：这道题目要对上面提到的位图稍作修改，这道题目中一个整数需要对应两个比特位。我们现有的位图都是一个整数对应一个比特位，自己再重新写一个对应两个比特位的会比较麻烦，这里我们可以直接使用现成的，开辟两个位图即可。

template<size_t N>
class twobitset
{
public:
  void set(size_t x)
  {
    //00-->01
    if (!_bs1.test(x) && !_bs2.test(x))
    {
      _bs2.set(x);
    }
    else if (!_bs1.test(x) && _bs2.test(x))//01-->10
    {
      _bs1.set(x);
      _bs2.reset(x);
    }
    //如果是10,那就表示出现次数是两次及以上
  }
  bool is_once(size_t x)
  {
    return !_bs1.test(x) && _bs2.test(x);
  }
private:
  bitset<N> _bs1;
  bitset<N> _bs2;
};
void Testtwobitset()
{
  int arr[] = { 1,1,2,3,77,7,7,7,90,100,100,90 };
  wcy::twobitset<101> tbs;
  for (auto e : arr)
  {
    tbs.set(e);
  }
  for (auto e : arr)
  {
    if (tbs.is_once(e))
    {
      cout << e << ' ';
    }
  }
}

1.4.2 给两个文件，分别有100亿个整数，我们只有1G内存，如何找到两个文件交集？

思路一：将其中一个文件的所有值映射到一个位图，然后去遍历另一个文件中的数据去判断在不在。这种思路存在一个缺陷，得到的交集中可能出现重复值的情况，而正常情况下交集中是不应该出现重复值，因此在前面求得交集后，还需要用 set 进行去重。这里可能会出现重复的数据过多的情况，去使用 set 可能会超过 1G 内存。

思路二： 将两个文件中的整数分别映射到两个位图中，然后再将两个位图按位与一下，得到一个新的位图，该位图中值为 1 的比特位对应的整数就是交集。

void Testintersection()
{
  int arr1[] = { 1,1,2,3,77,7,7,7,90,100,100,90 };
  int arr2[] = { 1111,12,2,31,771,71,7,70,901,100,10011,90 };
  wcy::bitset<10012> bs1;
  wcy::bitset<10012> bs2;
  for (auto e : arr1)
  {
    bs1.set(e);
  }
  for (auto e : arr2)
  {
    bs2.set(e);
  }
  for (size_t i = 0; i < 10012; ++i)
  {
    if (bs1.test(i) && bs2.test(i))
    {
      cout << i << ' ';
    }
  }
}

小Tips：上面用两个数组 arr1 和 arr2 来模拟两个文件。另外需要注意：位图到底需要多少个比特位不是取决于数据的个数，而是取决于这一堆数据可能的范围，数据范围有多大位图就需要开辟多少个比特位。

1.4.3 1个文件有100亿个int，1G内存，设计算法找到出现次数不超过2次的所有整数

思路：这道题的思路和 1.4.1 的思路是相同的，用两个位图，产生两个标记位来解决即可，这里就不再过多赘述。

二、布隆过滤器

2.1 布隆过滤器的提出

我们在使用新闻客户端看新闻时，它会给我们不停地推荐新的内容，它每次推荐时要去重，去掉那些已经看过的内容。问题来了，新闻客户端推荐系统如何实现推送去重的？用服务器记录了用户看过的所有历史记录，当推荐系统推荐新闻时会从每个用户的历史记录里进行筛选，过滤掉那些已经存在的记录。如何快速查找呢？

• 用哈希表存储用户记录，缺点：浪费空间。
  
• 用位图存储用户记录，缺点：位图一般只能处理整型，如果内容编号是字符串，就无法处理了。
  
• 将哈希与位图结合，即布隆过滤器。
  

2.2 布隆过滤器的概念

布隆过滤器是由布隆（Burton Howard Bloom）在 1970 年提出的一种紧凑型的、比较巧妙的概率型数据结构特点是高效地插入和查询，可以用来告诉你“某样东西一定不存在或者可能存在”，它是用多个哈希函数，将一个数据映射到位图结构中。此种方式不仅可以提升查询效率，也可以节省大量的内存空间。

小Tips：布隆过滤器如果判断出不再那么一定是不再，如果判断出在那么有可能是误判。因为在或不再分别是用 1 和 0 来标记的，如果不再那么标记位为 0，一定是真的。如果在，说明标记位是 1，这个 1 并不可信，因为这个 1 有可能是其它数据映射到这里来的，将该比特位置为 1。这个问题无法完全解决，只能通过增加哈希函数，让一个数据映射多个比特位，来尽最大的努力避免冲突发生，依此降低误判率。这里给大家分享一篇关于布隆过滤器的文章：详解布隆过滤器的原理，使用场景和注意事项。感兴趣的小伙伴可以去看看。

2.3 布隆过滤器的插入

//Bloom.h
struct BKDRHash
{
    size_t operator()(const string& str)
    {
        register size_t hash = 0;
        for (auto e : str)
        {
            hash = hash * 131 + e;   // 也可以乘以31、131、1313、13131、131313..  
            // 有人说将乘法分解为位运算及加减法可以提高效率，如将上式表达为：hash = hash << 7 + hash << 1 + hash + ch;  
            // 但其实在Intel平台上，CPU内部对二者的处理效率都是差不多的，  
            // 我分别进行了100亿次的上述两种运算，发现二者时间差距基本为0（如果是Debug版，分解成位运算后的耗时还要高1/3）；  
            // 在ARM这类RISC系统上没有测试过，由于ARM内部使用Booth's Algorithm来模拟32位整数乘法运算，它的效率与乘数有关：  
            // 当乘数8-31位都为1或0时，需要1个时钟周期  
            // 当乘数16-31位都为1或0时，需要2个时钟周期  
            // 当乘数24-31位都为1或0时，需要3个时钟周期  
            // 否则，需要4个时钟周期  
            // 因此，虽然我没有实际测试，但是我依然认为二者效率上差别不大          
        }
        return hash;
    }
};
struct APHash
{
    size_t operator()(const string& str)
    {
        register size_t hash = 0;
        for (size_t i = 0; i < str.size(); i++)
        {
            size_t ch = str[i];
            if ((i & 1) == 0)
            {
                hash ^= ((hash << 7) ^ ch ^ (hash >> 3));
            }
            else
            {
                hash ^= (~((hash << 11) ^ ch ^ (hash >> 5)));
            }
        }
        return hash;
    }
};
struct DJBHash
{
    size_t operator()(const string& str)
    {
        register size_t hash = 5381;
        for (auto ch : str)
        {
            hash += (hash << 5) + ch;
        }
        return hash;
    }
};
namespace wcy
{
  template<size_t N, class K = string
        ,class HashFunc1 = BKDRHash
        ,class HashFunc2 = APHash
        ,class HashFunc3 = DJBHash>
  class bloomfilter
  {
  public:
    void set(const K& key)
    {
            size_t hash1 = HashFunc1()(key) % N;
            _bs.set(hash1);
            size_t hash2 = HashFunc2()(key) % N;
            _bs.set(hash2);
            size_t hash3 = HashFunc3()(key) % N;
            _bs.set(hash3);
            cout << hash1 << ' ' << hash2 << ' ' << hash3 << endl;
    }
        bool test(const K& key)
        {
            size_t hash1 = HashFunc1()(key) % N;
            size_t hash2 = HashFunc2()(key) % N;
            size_t hash3 = HashFunc3()(key) % N;
            if (!_bs.test(hash1) || !_bs.test(hash2) || !_bs.test(hash3))
            {
                return false;//100%准确
            }
            return true;//存在误判
        }
  private:
    bitset<N> _bs;
  };
}

//test.cpp
void Test()
{
  wcy::bloomfilter<100> bf;
  bf.set("孙悟空");
  bf.set("牛魔王");
  bf.set("猪八戒");
  bf.set("二郎神");
  cout << bf.test("孙悟空") << endl;
  cout << bf.test("猪八戒") << endl;
  cout << bf.test("沙悟净") << endl;
}
int main()
{
  Test();
  return 0;
}

小Tips：由上面的两次运行结果可以看出，影响布隆过滤器的一个重要因素就是 N 的大小，这里的 N 表示底层位图要开辟 N 个比特位。N 越大意味着比特位越多，冲突的概率就会下降，布隆过滤器的误判率就会跟着下降，但是空间利用率也会下降。相反，N 越小空间利用率就会提升，但是冲突就会越多，布隆过滤器的误判率就会上升。那 N 到底多大合适呢？在程序中我们可以利用哈希中的那一套理论，即通过负载因子去控制 N 的大小。

小Tips：上图中 k 为哈希函数个数，m 为布隆过滤器长度，n 为插入的元素个数，p 为误判率。通过上图可以发现，影响布隆过滤器误判率的最主要因素还是哈希函数的个数，在增加了哈希函数的个数后，误判率有了明显的下降。

2.4 布隆过滤器的查找

布隆过滤器的思想是将一个元素用多个哈希函数映射到一个位图中，因此被映射到的位置的比特位一定为 1。所以可以按照以下方式进行查找：首先计算出每个哈希函数对应的哈希值，再去查看该哈希值对应的比特位置存储的是否为零，只要有一个为零，代表该元素一定不在布隆过滤器中，否则可能在布隆过滤器中。

注意：布隆过滤器如果说某个元素不在时，该元素一定不存在，如果布隆过滤器说该元素存在时，那么该元素可能存在，因为有一些哈希函数值存在一定的冲突，导致最终的误判。

例如：在布隆过滤器中查找 “沙悟净”时，假设 3 个哈希函数计算的哈希值为：1、3、7，刚好和其他元素对应的比特位重叠，此时布隆过滤器告诉该元素存在，但实际该元素是不存在的。

2.5 布隆过滤器的删除

布隆过滤器不能直接支持删除操作，因为在删除一个元素时，可能会影响其它元素。

例如：删除删除上面程序中的 “猪八戒”，如果直接将该元素所对应的二进制比特位置 0，那么 “二郎神” 也被删除了，因为这两个元素经过 HashFunc1 计算出来的结果是一样的，这就导致他俩对应的比特位会发生重叠。

一种正确的删除方法：将布隆过滤器中的每个比特位扩展成一个小的计数器，插入元素时给 k 个计数器（k个哈希函数计算出的哈希地址）加一，删除元素时，给 k 个计数器减一，通过多占用几倍存储空间的代价来给布隆过滤器增加删除操作。

缺陷：

• 无法确认元素是否真正在布隆过滤器中
  
• 存在计数回绕
  

2.6 布隆过滤器的优点

• 添加和查询的元素的时间复杂度是：O ( K ) O(K)O(K)（K 是哈希函数的个数，一般比较小），与数据量大小无关。
  
• 哈希函数相互之间没有关系，方便硬件并行运算。
  
• 布隆过滤器不需要存储元素本身，在某些对保密要求比较严格的场合有很大优势。
  
• 在能够承受一定的误判时，布隆过滤器与其他数据结构相比有着很大的空间优势。
  
• 在数据量很大的时候，布隆过滤器可以表示全集，其他数据结构不能。
  
• 使用同一组哈希函数的布隆过滤器可以进行交、并、差运算。
  

2.7 布隆过滤器的缺陷

• 有误判率，即存在假阳性（False Position），即不能准确判断元素是否在集合中（补数方法：再建立一个白名单，存储可能会误判的数据）。
  
• 不能获取元素本身。
  
• 一般情况下不能从布隆过滤器中删除元素。
  
• 如果采用计数方式删除，可能会存在计数回绕的问题。
  

2.8 布隆过滤器的实际应用场景

不需要精确的场景，比如快速判断昵称是否注册过。用户在注册账号时，输入一个昵称，在后端的数据库中要能够快速判断出该昵称是否已经存在数据库中，若存在说明该昵称已经被其他人注册过了。此时，就可以将数据库中的所有昵称都放到布隆过滤器中，若布隆过滤器判断出不在，那一定是不在，若布隆过滤器判断出在那可能是误判，我们可以通过负载因子来控制误判率，且这种误判影响不大。如果要精确判断，可以对上面的方式进行优化，若布隆过滤器判断出不在，那此结果一定是准确的，直接返回即可，若布隆过滤器判断的结果是存在，此时可以拿着该昵称到数据库中去搜索一遍，以数据库的查询结果返回给用户。

三、哈希切分

题目：给两个文件 A 和 B，分别有100亿个query，我们只有1G内存，如何找到两个文件交集？分别给出精确算法和近似算法。

思路一：首先假设 一个 query 占用 30Byte，那么100亿个 query 就会占用3000亿Byte，1G近似10亿Byte，所以3000亿Byte差不多就是300G。这里直接去遍历 A 中的元素再到 B 中去查找是不现实的，我们可以将这两个大文件分成若干份小文件，这里就将每个文件按照大小平均分成1000个小文件（A0~A999、B0~B999），这样以来每个小文件的大小就是300MB，然后我们可以依次拿着A0~A999这1000个小文件，去和B0~B999这1000个小文件进行对比，有相同的就放进交集。但是这样做的问题在于，A0~A999的每一个小文件都要和B的所有小文件对比一遍，这样效率也十分的低。所以下面我们引出哈希切分的方法。

思路二（哈希切分）：大思路上还是将 A 和 B 这两个大文件进行切分。但是并不再按照上面的内存大小进行平均切分，而是采用哈希切分。所谓哈希切分就是：根据一个哈希函数去计算确定将该元素划分到哪一个小文件里，可以采取下面这个公式进行计算：h a s h i = H a s h F u n c ( q u e r y ) hashi = HashFunc(query)hashi=HashFunc(query) % 1000 10001000。最终每个文件中存放的都是 hashi 相同的元素，这里的一个小文件就类似于一个哈希桶，里面的这些元素只有两种可能：重复、冲突。将 A 和 B 文件都采取相同的方式进行划分，最终 A 和 B 中相同的 query 一定会分别进入 Ai 和 Bi 编号相同的小文件。接下来我们只需要去对比 Ai 和 Bi 即可，这样效率会大大提高。这种做法还是存在一些小问题，有可能出现一个小文件中数据过多的情况，这样就不满足题目中的内存要求了。一个小文件中数据过多有以下两种情形：

• 比如 Ai 有 5G 数据，其中有 4G 都是相同的 query（即重复数据），还有 1G 是不同的 query（即冲突数据）。
  
• 基本上都是不同的数据（即冲突数据）。
  

针对这两种情形提出以下解决方案：

• 先把 Ai 的 query 读到一个 set 里面，如果 set 的 insert 报错抛异常（bad_alloc，内存错误），那么就说明该小文件中大多数 query 都是不相同的（即冲突）。如果能够全部读出来，insert 到 set 里面，那么说明 Ai 里面大部分都是相同的 query（这里利用了 set 可以去重的机制）
  
• 如果抛异常，说明有大量冲突，就再换一个哈希函数，进行二次切分。
  

小Tips：这道题目和 1.4.2 不同。这道题中的 query 不是整数，不适合使用位图来解决。使用哈希切分的目的就是为了满足内存限制，切分后再使用 set，对小文件的数据做进一步处理，如果小文件中都是重复的数据，set 会起到去重的作用，如果小文件中都是不相同的冲突数据，set 会抛异常，那么我们就再找一个哈希函数对该小文件继续切分。最终再用 set 去查找在不在，这样还可以做到精确查找。要做近似查找，直接使用布隆过滤器即可。

题目：给一个超过100G大小的log file, log中存着IP地址, 设计算法找到出现次数最多的IP地址？

思路：这道题还是采取哈希切分的方法去完成。用一个哈希函数将 100G 大小的文件分成若干份小文件，相同的 IP 一定会被划分到同一个小文件中，然后再用一个 map 去统计出每一个小文件中 IP 出现的次数。记录下出现次数最多的即可。

四、结语

今天的分享到这里就结束啦！如果觉得文章还不错的话，可以三连支持一下，春人的主页还有很多有趣的文章，欢迎小伙伴们前去点评，您的支持就是春人前进的动力！