分布式爬虫-基于redis的布隆过滤器设计大规模网址去重

Posted 2021-05-01 吞宇闲聊AI

当前浏览器不支持播放音乐或语音，请在微信或其他浏览器中播放

利用Redis的Bitmap数据类型做布隆过滤器，基于redis实现的分布式锁来保证数据一致性。

一、布隆过滤器应用场景：

检查一个英语单词是否拼写正确；判断一个嫌疑人是否在嫌疑人名单中；判断一个网址是否被访问过等。

优势：占用空间小、查询速度快。

缺点：有一定的误判。

二、布隆过滤器

布隆过滤器是一个很长的二进制向量和一系列随机映射函数。

原理：底层使用的是位图。当一个元素被加入集合时，通过 K 个 Hash 函数将这个元素映射成一个位阵列（Bit array）中的 K 个点，把它们置为 1。检索时，我们只要看看这些点是不是都是 1 就（大约）知道集合中有没有它了：

1、如果这些点有任何一个 0，则被检索元素一定不在；

2、如果都是 1，则被检索元素很可能在。

三、Redis

Redis 因其支持 setbit 和 getbit 操作，且纯内存性能高等特点，因此天然就可以作为布隆过滤器来使用。根据《数学之美》中给出的数据，在使用8个哈希函数的情况下，512MB大小的位数组在误报率万分之五的情况下可以对约两亿的url去重。

四、分布式锁

在多线程下，同时对某一个共享数据进行读写，会出现脏数据、数据不一致现象。在分布式式系统中，分布式锁可用于保持数据一致性。分布式锁实现方式有基于数据库实现、基于缓存实现、基于zookeeper实现。本文选择基于redis缓存实现分布式锁。主要原因如下：

第一、  redis是基于内存来操作，存取速度比数据库快，在高并发下，加锁之后的性能不会下降太多。

第二、  redis 可以设置键值的生存时间（TTL）

第三、  redis 的使用方式简单，总体实现开销小

五、核心代码示例

1）通过redis加锁

    public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {

        String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
        jedis.setex(lockKey, expireTime, requestId);
//        RedisUtil.returnResource(jedis);
        if (LOCK_SUCCESS.equals(result)) {
            return true;
        }
        return false;
    }

2）释放锁

    public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {

        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
//        RedisUtil.returnResource(jedis);
        if (RELEASE_SUCCESS.equals(result)) {
            return true;
        }
        return false;
    }

3）判断元素是否存在，如果存在则返回true，如果不存在则将布隆过滤器置为1，返回false。

    public boolean contains(String key) {
        Boolean result = true;
        // TODO 如果是集群模式这里需要分布式锁，如果是单机这里需要线程锁
        try {
            String lockKey = "lockKey";
            String requestId = UUID.randomUUID().toString();
            int expireTime = 10;
            Boolean b = RedisTool.tryGetDistributedLock(jedis, lockKey, requestId, expireTime);
            if (b) {
                Objects.requireNonNull(key, "the key must not be null");
                int hash = getHash(key);
                int offset = hash % fixSize;
                String bitKey = getBitKey(hash);
                result = jedis.getbit(bitKey, offset);
                jedis.setbit(bitKey, offset, true);
                RedisTool.releaseDistributedLock(jedis, lockKey, requestId);
            }
            RedisTool.releaseDistributedLock(jedis, lockKey, requestId);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;
    }

4）布隆过滤器工具类完整代码

package toolkit;

import java.util.Objects;
import java.util.UUID;
import redis.clients.jedis.Jedis;

public class BloomFilterUtil {
    private String nameSpace;
    private Jedis jedis;
    private int fixSize;

    public BloomFilterUtil(String nameSpace, Jedis jedis, int fixSize) {
        this.nameSpace = nameSpace;
        this.jedis = jedis;
        this.fixSize = fixSize;
    }

    private int getHash(String key) {
        return key.hashCode()& Integer.MAX_VALUE;
    }

    /**
     * 根据hash和fixSize 算出bitKey
     * 
     * @param hash
     * @return
     */
    private String getBitKey(int hash) {
        String bitKey = nameSpace;
        return bitKey;
    }

    /**
     * 判断给定的key是否存在
     * 
     * @param key
     * @return
     */
    public boolean contains(String key) {
        Boolean result = true;
        // TODO 如果是集群模式这里需要分布式锁，如果是单机这里需要线程锁
        try {
            String lockKey = "lockKey";
            String requestId = UUID.randomUUID().toString();
            int expireTime = 10;
            Boolean b = RedisTool.tryGetDistributedLock(jedis, lockKey, requestId, expireTime);
            if (b) {
                Objects.requireNonNull(key, "the key must not be null");
                int hash = getHash(key);
                int offset = hash % fixSize;
                String bitKey = getBitKey(hash);
                result = jedis.getbit(bitKey, offset);
                jedis.setbit(bitKey, offset, true);
                RedisTool.releaseDistributedLock(jedis, lockKey, requestId);
            }
            RedisTool.releaseDistributedLock(jedis, lockKey, requestId);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;
    }


    public static void main(String[] args) {
        String nameSpace = "bloomfilter";
        Jedis jedis = RedisUtil.putJedis();
        int fixSize = 100000000;
        BloomFilterUtil rbAndDbBloomFilter = new BloomFilterUtil(nameSpace, jedis, fixSize);
        System.out.println(rbAndDbBloomFilter.contains("b"));
        for(int i = 1;i<10;i++) {
            System.out.println(rbAndDbBloomFilter.contains("https://cangzhou.focus.cn/loupan/q167/"+i));
        }
    }
}

六、运行效果

第一遍，元素都不存在，结果都是false。

false
false
false
false

第二遍，元素已经存在，结果都是true。

true
true
true
true