布隆過濾器實現代碼php+redis

摘自：http://imhuchao.com/1271.html

引言

在介紹布隆過濾器之前我們首先引入幾個場景。

場景一

在一個高並發的計數系統中，如果一個key沒有計數，此時我們應該返回0，但是訪問的key不存在，相當於每次訪問緩存都不起作用了。那么如何避免頻繁訪問數量為0的key而導致的緩存被擊穿？

有人說， 將這個key的值置為0存入緩存不就行了嗎？確實，這是一個好的方案。大部分情況我們都是這樣做的，當訪問一個不存在的key的時候，設置一個帶有過期時間的標志，然后放入緩存。不過這樣做的缺點也很明顯，浪費內存和無法抵御隨機key攻擊。

場景二

在一個黑名單系統中，我們需要設置很多黑名單內容。比如一個郵件系統，我們需要設置黑名單用戶，當判斷垃圾郵件的時候，要怎么去做。比如爬蟲系統，我們要記錄下來已經訪問過的鏈接避免下次訪問重復的鏈接。

在郵件很少或者用戶很少的情況下，我們用普通數據庫自帶的查詢就能完成。在數據量太多的時候，為了保證速度，通常情況下我們會將結果緩存到內存中，數據結構用hash表。這種查找的速度是O(1)，但是內存消耗也是驚人的。打個比方，假如我們要存10億條數據，每條數據平均占據32個字節，那么需要的內存是64G，這已經是一個驚人的大小了。

一種解決思路

能不能有一種思路，查詢的速度是O(1)，消耗內存特別小呢？前輩門早就想出了一個很好的解決方案。由於上面說的場景判斷的結果只有兩種狀態（是或者不是，存在或者不存在），那么對於所存的數據完全可以用位來表示！數據本身則可以通過一個hash函數計算出一個key，這個key是一個位置，而這個key所對的值就是0或者1（因為只有兩種狀態），如下圖：

 

布隆過濾器原理

上面的思路其實就是布隆過濾器的思想，只不過因為hash函數的限制，多個字符串很可能會hash成一個值。為了解決這個問題，布隆過濾器引入多個hash函數來降低誤判率。

下圖表示有三個hash函數，比如一個集合中有x，y，z三個元素，分別用三個hash函數映射到二進制序列的某些位上，假設我們判斷w是否在集合中，同樣用三個hash函數來映射，結果發現取得的結果不全為1，則表示w不在集合里面。

布隆過濾器處理流程

布隆過濾器應用很廣泛，比如垃圾郵件過濾，爬蟲的url過濾，防止緩存擊穿等等。下面就來說說布隆過濾器的一個完整流程，相信讀者看到這里應該能明白布隆過濾器是怎樣工作的。

第一步：開辟空間

開辟一個長度為m的位數組（或者稱二進制向量），這個不同的語言有不同的實現方式，甚至你可以用文件來實現。

第二步：尋找hash函數

獲取幾個hash函數，前輩們已經發明了很多運行良好的hash函數，比如BKDRHash，JSHash，RSHash等等。這些hash函數我們直接獲取就可以了。

第三步：寫入數據

將所需要判斷的內容經過這些hash函數計算，得到幾個值，比如用3個hash函數，得到值分別是1000，2000，3000。之后設置m位數組的第1000，2000，3000位的值位二進制1。

第四步：判斷

接下來就可以判斷一個新的內容是不是在我們的集合中。判斷的流程和寫入的流程是一致的。

誤判問題

布隆過濾器雖然很高效（寫入和判斷都是O(1)，所需要的存儲空間極小），但是缺點也非常明顯，那就是會誤判。當集合中的元素越來越多，二進制序列中的1的個數越來越多的時候，判斷一個字符串是否在集合中就很容易誤判，原本不在集合里面的字符串會被判斷在集合里面。

數學推導

布隆過濾器原理十分簡單，但是hash函數個數怎么去判斷，誤判率有多少？

假設二進制序列有m位，那么經過當一個字符串hash到某一位的概率為：

1𝑚

也就是說當前位被反轉為1的概率：

𝑝(1)=1𝑚

那么這一位沒有被反轉的概率為：

𝑝(0)=1−1𝑚

假設我們存入n各元素，使用k個hash函數，此時沒有被翻轉的概率為：

𝑝(0)=(1−1𝑚)𝑛𝑘

那什么情況下我們會誤判呢，就是原本不應該被翻轉的位，結果翻轉了，也就是

𝑝(誤判)=1−(1−1𝑚)𝑛𝑘

由於只有k個hash函數同時誤判了，整體才會被誤判，最后誤判的概率為

𝑝(誤判)=(1−(1−1𝑚)𝑛𝑘)𝑘

要使得誤判率最低，那么我們需要求誤判與m、n、k之間的關系，現在假設m和n固定，我們計算一下k。可以首先看看這個式子：

(1−1𝑚)𝑛𝑘

由於我們的m很大，通常情況下我們會用2^32來作為m的值。上面的式子中含有一個重要極限

lim𝑥→∞(1+1𝑥)𝑥=𝑒

因此誤判率的式子可以寫成

 𝑝(誤判)=(1−(𝑒)−𝑛𝑘/𝑚)𝑘

接下來令𝑡=−𝑛/𝑚，兩邊同時取對數，求導，得到：

𝑝′1𝑝=𝑙𝑛(1−𝑒𝑡𝑘)+𝑘𝑙𝑛𝑒𝑡(−𝑒𝑡𝑘)1−𝑒𝑡𝑘

讓𝑝′=0，則等式后面的為0，最后整理出來的結果是

(1−𝑒𝑡𝑘)𝑙𝑛(1−𝑒𝑡𝑘)=𝑒𝑡𝑘𝑙𝑛𝑒𝑡𝑘

計算出來的k為𝑙𝑛2𝑚𝑛，約等於0.693𝑚𝑛，將k代入p(誤判)，我們可以得到概率和m、n之間的關系，最后的結果

(1/2)𝑙𝑛2𝑚𝑛，約等於0.6185𝑚/𝑛

以上我們就得出了最佳hash函數個數以及誤判率與mn之前的關系了。

下表是m與n比值在k個hash函數下面的誤判率

m/n	k	k=1	k=2	k=3	k=4	k=5	k=6	k=7	k=8
2	1.39	0.393	0.400
3	2.08	0.283	0.237	0.253
4	2.77	0.221	0.155	0.147	0.160
5	3.46	0.181	0.109	0.092	0.092	0.101
6	4.16	0.154	0.0804	0.0609	0.0561	0.0578	0.0638
7	4.85	0.133	0.0618	0.0423	0.0359	0.0347	0.0364
8	5.55	0.118	0.0489	0.0306	0.024	0.0217	0.0216	0.0229
9	6.24	0.105	0.0397	0.0228	0.0166	0.0141	0.0133	0.0135	0.0145
10	6.93	0.0952	0.0329	0.0174	0.0118	0.00943	0.00844	0.00819	0.00846
11	7.62	0.0869	0.0276	0.0136	0.00864	0.0065	0.00552	0.00513	0.00509
12	8.32	0.08	0.0236	0.0108	0.00646	0.00459	0.00371	0.00329	0.00314
13	9.01	0.074	0.0203	0.00875	0.00492	0.00332	0.00255	0.00217	0.00199
14	9.7	0.0689	0.0177	0.00718	0.00381	0.00244	0.00179	0.00146	0.00129
15	10.4	0.0645	0.0156	0.00596	0.003	0.00183	0.00128	0.001	0.000852
16	11.1	0.0606	0.0138	0.005	0.00239	0.00139	0.000935	0.000702	0.000574
17	11.8	0.0571	0.0123	0.00423	0.00193	0.00107	0.000692	0.000499	0.000394
18	12.5	0.054	0.0111	0.00362	0.00158	0.000839	0.000519	0.00036	0.000275
19	13.2	0.0513	0.00998	0.00312	0.0013	0.000663	0.000394	0.000264	0.000194
20	13.9	0.0488	0.00906	0.0027	0.00108	0.00053	0.000303	0.000196	0.00014
21	14.6	0.0465	0.00825	0.00236	0.000905	0.000427	0.000236	0.000147	0.000101
22	15.2	0.0444	0.00755	0.00207	0.000764	0.000347	0.000185	0.000112	7.46e-05
23	15.9	0.0425	0.00694	0.00183	0.000649	0.000285	0.000147	8.56e-05	5.55e-05
24	16.6	0.0408	0.00639	0.00162	0.000555	0.000235	0.000117	6.63e-05	4.17e-05
25	17.3	0.0392	0.00591	0.00145	0.000478	0.000196	9.44e-05	5.18e-05	3.16e-05
26	18	0.0377	0.00548	0.00129	0.000413	0.000164	7.66e-05	4.08e-05	2.42e-05
27	18.7	0.0364	0.0051	0.00116	0.000359	0.000138	6.26e-05	3.24e-05	1.87e-05
28	19.4	0.0351	0.00475	0.00105	0.000314	0.000117	5.15e-05	2.59e-05	1.46e-05
29	20.1	0.0339	0.00444	0.000949	0.000276	9.96e-05	4.26e-05	2.09e-05	1.14e-05
30	20.8	0.0328	0.00416	0.000862	0.000243	8.53e-05	3.55e-05	1.69e-05	9.01e-06
31	21.5	0.0317	0.0039	0.000785	0.000215	7.33e-05	2.97e-05	1.38e-05	7.16e-06
32	22.2	0.0308	0.00367	0.000717	0.000191	6.33e-05	2.5e-05	1.13e-05	5.73e-06

php+Redis實現的布隆過濾器

由於Redis實現了setbit和getbit操作，天然適合實現布隆過濾器，redis也有布隆過濾器插件。這里使用php+redis實現布隆過濾器。

首先定義一個hash函數集合類，這些hash函數不一定都用到，實際上32位hash值的用3個就可以了，具體的數量可以根據你的位序列總量和你需要存入的量決定，上面已經給出最佳值。

class BloomFilterHash
{
    /**
     * 由Justin Sobel編寫的按位散列函數
     */
    public function JSHash($string, $len = null)
    {
        $hash = 1315423911;
        $len || $len = strlen($string);
        for ($i=0; $i<$len; $i++) {
            $hash ^= (($hash << 5) + ord($string[$i]) + ($hash >> 2));
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * 該哈希算法基於AT＆T貝爾實驗室的Peter J. Weinberger的工作。
     * Aho Sethi和Ulman編寫的“編譯器（原理，技術和工具）”一書建議使用采用此特定算法中的散列方法的散列函數。
     */
    public function PJWHash($string, $len = null)
    {
        $bitsInUnsignedInt = 4 * 8; //（unsigned int）（sizeof（unsigned int）* 8）;
        $threeQuarters = ($bitsInUnsignedInt * 3) / 4;
        $oneEighth = $bitsInUnsignedInt / 8;
        $highBits = 0xFFFFFFFF << (int) ($bitsInUnsignedInt - $oneEighth);
        $hash = 0;
        $test = 0;
        $len || $len = strlen($string);
        for($i=0; $i<$len; $i++) {
            $hash = ($hash << (int) ($oneEighth)) + ord($string[$i]); } $test = $hash & $highBits; if ($test != 0) { $hash = (($hash ^ ($test >> (int)($threeQuarters))) & (~$highBits));
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * 類似於PJW Hash功能，但針對32位處理器進行了調整。它是基於UNIX的系統上的widley使用哈希函數。
     */
    public function ELFHash($string, $len = null)
    {
        $hash = 0;
        $len || $len = strlen($string);
        for ($i=0; $i<$len; $i++) {
            $hash = ($hash << 4) + ord($string[$i]); $x = $hash & 0xF0000000; if ($x != 0) { $hash ^= ($x >> 24);
            }
            $hash &= ~$x;
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * 這個哈希函數來自Brian Kernighan和Dennis Ritchie的書“The C Programming Language”。
     * 它是一個簡單的哈希函數，使用一組奇怪的可能種子，它們都構成了31 .... 31 ... 31等模式，它似乎與DJB哈希函數非常相似。
     */
    public function BKDRHash($string, $len = null)
    {
        $seed = 131;  # 31 131 1313 13131 131313 etc..
        $hash = 0;
        $len || $len = strlen($string);
        for ($i=0; $i<$len; $i++) {
            $hash = (int) (($hash * $seed) + ord($string[$i]));
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * 這是在開源SDBM項目中使用的首選算法。
     * 哈希函數似乎對許多不同的數據集具有良好的總體分布。它似乎適用於數據集中元素的MSB存在高差異的情況。
     */
    public function SDBMHash($string, $len = null)
    {
        $hash = 0;
        $len || $len = strlen($string);
        for ($i=0; $i<$len; $i++) {
            $hash = (int) (ord($string[$i]) + ($hash << 6) + ($hash << 16) - $hash);
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * 由Daniel J. Bernstein教授制作的算法，首先在usenet新聞組comp.lang.c上向世界展示。
     * 它是有史以來發布的最有效的哈希函數之一。
     */
    public function DJBHash($string, $len = null)
    {
        $hash = 5381;
        $len || $len = strlen($string);
        for ($i=0; $i<$len; $i++) {
            $hash = (int) (($hash << 5) + $hash) + ord($string[$i]);
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * Donald E. Knuth在“計算機編程藝術第3卷”中提出的算法，主題是排序和搜索第6.4章。
     */
    public function DEKHash($string, $len = null)
    {
        $len || $len = strlen($string);
        $hash = $len;
        for ($i=0; $i<$len; $i++) {
            $hash = (($hash << 5) ^ ($hash >> 27)) ^ ord($string[$i]);
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }

    /**
     * 參考 http://www.isthe.com/chongo/tech/comp/fnv/
     */
    public function FNVHash($string, $len = null)
    {
        $prime = 16777619; //32位的prime 2^24 + 2^8 + 0x93 = 16777619
        $hash = 2166136261; //32位的offset
        $len || $len = strlen($string);
        for ($i=0; $i<$len; $i++) {
            $hash = (int) ($hash * $prime) % 0xFFFFFFFF;
            $hash ^= ord($string[$i]);
        }
        return ($hash % 0xFFFFFFFF) & 0xFFFFFFFF;
    }
}

 

接着就是連接redis來進行操作

/**
 * 使用redis實現的布隆過濾器
 */
abstract class BloomFilterRedis
{
    /**
     * 需要使用一個方法來定義bucket的名字
     */
    protected $bucket;

    protected $hashFunction;

    public function __construct($config, $id)
    {
        if (!$this->bucket || !$this->hashFunction) {
            throw new Exception("需要定義bucket和hashFunction", 1);
        }
        $this->Hash = new BloomFilterHash;
        $this->Redis = new YourRedis; //假設這里你已經連接好了
    }

    /**
     * 添加到集合中
     */
    public function add($string)
    {
        $pipe = $this->Redis->multi();
        foreach ($this->hashFunction as $function) {
            $hash = $this->Hash->$function($string);
            $pipe->setBit($this->bucket, $hash, 1);
        }
        return $pipe->exec();
    }

    /**
     * 查詢是否存在, 存在的一定會存在, 不存在有一定幾率會誤判
     */
    public function exists($string)
    {
        $pipe = $this->Redis->multi();
        $len = strlen($string);
        foreach ($this->hashFunction as $function) {
            $hash = $this->Hash->$function($string, $len);
            $pipe = $pipe->getBit($this->bucket, $hash);
        }
        $res = $pipe->exec();
        foreach ($res as $bit) {
            if ($bit == 0) {
                return false;
            }
        }
        return true;
    }

}

 

上面定義的是一個抽象類，如果要使用，可以根據具體的業務來使用。比如下面是一個過濾重復內容的過濾器。

/**
 * 重復內容過濾器
 * 該布隆過濾器總位數為2^32位, 判斷條數為2^30條. hash函數最優為3個.(能夠容忍最多的hash函數個數)
 * 使用的三個hash函數為
 * BKDR, SDBM, JSHash
 *
 * 注意, 在存儲的數據量到2^30條時候, 誤判率會急劇增加, 因此需要定時判斷過濾器中的位為1的的數量是否超過50%, 超過則需要清空.
 */
class FilteRepeatedComments extends BloomFilterRedis
{
    /**
     * 表示判斷重復內容的過濾器
     * @var string
     */
    protected $bucket = 'rptc';

    protected $hashFunction = array('BKDRHash', 'SDBMHash', 'JSHash');
}