Redis 布隆过滤器实战「缓存击穿、雪崩效应」

$ python2
Python 2.7.10 (default, Oct  6 2017, 22:29:07)
[GCC 4.2.1 Compatible Apple LLVM 9.0.0 (clang-900.0.31)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> s = {2, 4}
>>> filter(lambda x:x in s, [0, 1, 2])
[2]
复制代码

2**128 = 340282366920938463463374607431768211456 bit
复制代码

import crc32
def BloomFilter(sample, size, hash_size=1):
    # 构造一个hash函数，将输入数据散列到size一个位置上
    hash = lambda x:crc32(str(x).encode())%size
    collision, s = 0, set()
    for i in range(sample):
        k = set()
        for j in range(hash_size):
            k.add(hash(i+j*size/hash_size))
        # 只有所有散列结果k都在s中，才认为i重复
        if not k - s:
            collision += 1
            continue
        # 将散列结果k更新到集合s中
        s |= k
    return collision
复制代码

pyreBloom:master λ ls
Makefile      bloom.h       bloom.pxd     murmur.c      pyreBloom.pyx
bloom.c       bloom.o       main.c        pyreBloom.c
复制代码

int init_pyrebloom(pyrebloomctxt * ctxt, char * key, uint32_t capacity, double error, char* host, uint32_t port, char* password, uint32_t db);
int free_pyrebloom(pyrebloomctxt * ctxt);
int add(pyrebloomctxt * ctxt, const char * data, uint32_t len);
int add_complete(pyrebloomctxt * ctxt, uint32_t count);
int check(pyrebloomctxt * ctxt, const char * data, uint32_t len);
int check_next(pyrebloomctxt * ctxt);
int delete(pyrebloomctxt * ctxt);
复制代码

import math
import random
cimport bloom
class pyreBloomException(Exception):
  '''Some sort of exception has happened internally'''
  pass
cdef class pyreBloom(object):
  cdef bloom.pyrebloomctxt context
  cdef bytes               key
  property bits:
    def __get__(self):
      return self.context.bits
  property hashes:
    def __get__(self):
      return self.context.hashes
  def __cinit__(self, key, capacity, error, host='127.0.0.1', port=6379,
    password='', db=0):
    self.key = key
    if bloom.init_pyrebloom(&self.context, self.key, capacity,
      error, host, port, password, db):
      raise pyreBloomException(self.context.ctxt.errstr)
  def __dealloc__(self):
    bloom.free_pyrebloom(&self.context)
  def delete(self):
    bloom.delete(&self.context)
  def put(self, value):
    if getattr(value, '__iter__', False):
      r = [bloom.add(&self.context, v, len(v)) for v in value]
      r = bloom.add_complete(&self.context, len(value))
    else:
      bloom.add(&self.context, value, len(value))
      r = bloom.add_complete(&self.context, 1)
    if r < 0:
      raise pyreBloomException(self.context.ctxt.errstr)
    return r
  def add(self, value):
    return self.put(value)
  def extend(self, values):
    return self.put(values)
  def contains(self, value):
    # If the object is 'iterable'...
    if getattr(value, '__iter__', False):
      r = [bloom.check(&self.context, v, len(v)) for v in value]
      r = [bloom.check_next(&self.context) for i in range(len(value))]
      if (min(r) < 0):
        raise pyreBloomException(self.context.ctxt.errstr)
      return [v for v, included in zip(value, r) if included]
    else:
      bloom.check(&self.context, value, len(value))
      r = bloom.check_next(&self.context)
      if (r < 0):
        raise pyreBloomException(self.context.ctxt.errstr)
      return bool(r)
  def __contains__(self, value):
    return self.contains(value)
  def keys(self):
    '''Return a list of the keys used in this bloom filter'''
    return [self.context.keys[i] for i in range(self.context.num_keys)]
复制代码

原生pyreBloom方法:
cdef class pyreBloom(object):
    cdef bloom.pyrebloomctxt context
    cdef bytes
    property bits:
    property hashes:
    // 使用的hash方法数
    def delete(self):
    // 删除，会在redis中删除
    def put(self, value):
    // 添加 底层方法, 不建议直接调用
    def add(self, value):
    // 添加单个元素，调用put方法
    def extend(self, values):
    // 添加一组元素，调用put方法
    def contains(self, value):
    // 检查是否存在，当`value`可以迭代时，返回`[value]`, 否则返回`bool`
    def keys(self):
    // 在redis中存储的key列表
复制代码

# coding=utf-8
    '''
    bloom filter 基础模块
    可用方法:
    extend, keys, contains, add, put, hashes, bits, delete
    使用方法:
    >>> class TestModel(BaseModel):
    ...    PREFIX = "bf:test"
    >>> t = TestModel()
    >>> t.add('hello')
    1
    >>> t.extend(['hi', 'world'])
    2
    >>> t.contains('hi')
    True
    >>> t.delete()
    '''
    import logging
    from six import PY3 as IS_PY3
    from pyreBloom import pyreBloom, pyreBloomException
    from BloomFilter.utils import force_utf8
    class BaseModel(object):
        '''
        bloom filter 基础模块
        参数：
            SLOT: 可用方法类型
            PREFIX: redis前缀
            BF_SIZE: 存储最大值
            BF_ERROR: 允许的出错率
            RETRIES: 连接重试次数
            host: redis 服务器IP
            port: redis 服务器端口
            db: redis 服务器DB
            _bf_conn: 内部保存`pyreBloom`实例
        '''
        SLOT = {'add', 'contains', 'extend', 'keys', 'put', 'delete',
                'bits', 'hashes'}
        PREFIX = ""
        BF_SIZE = 100000
        BF_ERROR = 0.01
        RETRIES = 2
        def __init__(self, redis=None):
            '''
            初始化redis配置
            :param redis: redis 配置
            '''
            # 这里初始化防止类静态变量多个继承类复用，导致数据被污染
            self._bf_conn = None
            self._conf = {
                'host': '127.0.0.1', 'password': '',
                'port': 6379, 'db': 0
            }
            if redis:
                for k, v in redis.items():
                    if k in self._conf:
                        self._conf[k] = redis[k]
            self._conf = force_utf8(self._conf)
        @property
        def bf_conn(self):
            '''
            初始化pyreBloom
            '''
            if not self._bf_conn:
                prefix = force_utf8(self.PREFIX)
                logging.debug(
                    'pyreBloom connect: redis://%s:%s/%s, (%s %s %s)',
                    self._conf['host'], self._conf['port'], self._conf['db'],
                    prefix, self.BF_SIZE, self.BF_ERROR,
                )
                self._bf_conn = pyreBloom(
                    prefix, self.BF_SIZE, self.BF_ERROR, **self._conf)
            return self._bf_conn
        def __getattr__(self, method):
            '''调用pyrebloom方法
            没有枚举的方法将从`pyreBloom`中获取
            :param method:
            :return: pyreBloom.{method}
            '''
            # 只提供内部方法
            if method not in self.SLOT:
                raise NotImplementedError()
            # 捕获`pyreBloom`的异常, 打印必要的日志
            def catch_error(*a, **kwargs):
                '''多次重试服务'''
                args = force_utf8(a)
                kwargs = force_utf8(kwargs)
                for _ in range(self.RETRIES):
                    try:
                        func = getattr(self.bf_conn, method)
                        res = func(*args, **kwargs)
                        # python3 返回值和python2返回值不相同，
                        # 手工处理返回类型
                        if method == 'contains' and IS_PY3:
                            if isinstance(res, list):
                                return [i.decode('utf8') for i in res]
                        return res
                    except pyreBloomException as error:
                        logging.warn(
                            'pyreBloom Error:  %s %s', method, str(error))
                        self.reconnect()
                        if _ == self.RETRIES:
                            logging.error('pyreBloom Error')
                            raise error
            return catch_error
        def __contains__(self, item):
            '''跳转__contains__方法
            :param item: 查询元素列表/单个元素
            :type item: list/basestring
            :return: [bool...]/bool
            '''
            return self.contains(item)
        def reconnect(self):
            '''
            重新连接bloom
            `pyreBloom` 连接使用c driver，没有提供timeout参数，使用了内置的timeout
            同时为了保证服务的可靠性，增加了多次重试机制。
            struct timeval timeout = { 1, 5000 };
            ctxt->ctxt = redisConnectWithTimeout(host, port, timeout);
            del self._bf_conn 会调用`pyreBloom`内置的C的del方法，会关闭redis连接
            '''
            if self._bf_conn:
                logging.debug('pyreBloom reconnect')
                del self._bf_conn
                self._bf_conn = None
                _ = self.bf_conn
复制代码