一、前言
在实际开发中,会遇到很多要判断一个元素是否在某个集合中的业务场景,类似于垃圾邮件的识别,恶意ip地址的访问,缓存穿透等情况。类似于缓存穿透这种情况,有许多的解决方法,如:redis存储null值等,而对于垃圾邮件的识别,恶意ip地址的访问,我们也可以直接用 HashMap 去存储恶意ip地址以及垃圾邮件,然后每次访问时去检索一下对应集合中是否有相同数据。这种思路对于数据量小的项目来说是没有问题的,但是对于大数据量的项目,如,垃圾邮件出现有十几二十万,恶意ip地址出现有上百万,或者从几十亿电话中检索出指定的电话是否在等操作,那么这十几亿的数据就会占据大几G的空间,这个时候就可以考虑一下布隆过滤器了。
二、布隆过滤器(Bloom Filter)
一、是什么
布隆过滤器(Bloom Filter)是1970年由布隆提出的。它实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。
一句话就是:由一个初始值为零的bit数组和多个哈希函数构成,用来快速判断集合中是否存在某个元素。
使用bit数组的目的就是减少内存的占用,数组不保存数据信息,只是在内存中存储一个是否存在的表示0或1
二、原理
一、原理
当一个元素被加入集合时,通过 K 个 Hash 函数将这个元素映射成一个位阵列(Bit array)中的 K 个点,把它们置为 1。检索时,我们只要看看这些点是不是都是 1 就(大约)知道集合中有没有它了。
1、添加key
使用多个hash函数对key进行hash运算得到多个整数索引值,对位数组长度进行取模运算得到多个位置,每个hash函数都会得到一个不同的位置,将这几个位置都置1就完成了add操作。
例如,我们添加一个字符串wmyskxz,对字符串进行多次hash(key) → 取模运行→ 得到坑位
2、查询key
将这个key的多个位置上的值取出来,只要有其中一位是零就表示这个key不存在,但如果都是1,则不一定存在对应的key。(也就是有,不一定有,无,就一定无)
比如我们在 add 了字符串wmyskxz数据之后,很明显下面1/3/5 这几个位置的 1 是因为第一次添加的 wmyskxz 而导致的;
此时我们查询一个没添加过的不存在的字符串inexistent-key,它有可能计算后坑位也是1/3/5 ,这就是误判了
二、hash冲突导致数据不精准
当有变量被加入集合时,通过N个映射函数将这个变量映射成位图中的N个点,
把它们置为 1(假定有两个变量都通过 3 个映射函数)。
为什么说有,不一定有,无,就一定无。那是因为映射函数本身就是散列函数,散列函数是会有碰撞的。如上图,obj1和obj2放入的位置都是相同的,如果只放入obj2不放入obj1,然后查key为obj1的也是能够查到bit数组上都是1的结果,但这并不代表obj1就存在。
1、hash函数
将任意大小的输入数据转换成特定大小的输出数据的函数,转换后的数据称为哈希值或哈希编码,也叫散列值。
如果两个散列值是不相同的(根据同一函数)那么这两个散列值的原始输入也是不相同的。
这个特性是散列函数具有确定性的结果,具有这种性质的散列函数称为单向散列函数。
散列函数的输入和输出不是唯一对应关系的,如果两个散列值相同,两个输入值很可能是相同的,但也可能不同,
这种情况称为“散列碰撞(collision)”。
用 hash表存储大数据量时,空间效率还是很低,当只有一个 hash 函数时,还很容易发生哈希碰撞。
2、hash冲突代码复现
package test;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
/**
* @Description : 哈希冲突复现
* @Author : hc
* @Date : 2023/6/14 19:02
**/
public class HashCodeTest {
public static void main(String[] args) {
Set<Integer> hashCodeSet = new HashSet<>();
for (int i = 0; i < 200000; i++) {
int hashCode = new Object().hashCode();
if (hashCodeSet.contains(hashCode)) {
System.out.println("出现了重复的hashcode: " + hashCode + "\t 运行到" + i);
break;
}
hashCodeSet.add(hashCode);
}
System.out.println("Aa".hashCode());
System.out.println("BB".hashCode());
System.out.println("柳柴".hashCode());
System.out.println("柴柕".hashCode());
}
}
结果:
出现了重复的hashcode: 2134400190 运行到105084
2112
2112
851553
851553
三、布隆过滤器删除问题
原因:
布隆过滤器的误判是指多个输入经过哈希之后在相同的bit位置1了,这样就无法判断究竟是哪个输入产生的,
因此误判的根源在于相同的 bit 位被多次映射且置 1。
导致结果:
这种情况也造成了布隆过滤器的删除问题,因为布隆过滤器的每一个 bit 并不是独占的,很有可能多个元素共享了某一位。
如果我们直接删除这一位的话,会影响其他的元素
特性:
布隆过滤器可以添加元素,但是不能删除元素。因为删掉元素会导致误判率增加。只能重构。
总结:
1、使用时进行布隆过滤器的初始化,一次性给够容量,不要让实际数量大于初始化数量,避免重构布隆过滤器。
2、如果实际数量大于初始化数量,这个时候就需要进行重构了,重新分配一个更大数量的过滤器,再将所有旧数据重新初始化进过滤器。
三、使用场景
一、黑白名单校验、识别垃圾邮件
发现存在黑名单中的,就执行特定操作。比如:识别垃圾邮件,只要是邮箱在黑名单中的邮件,就识别为垃圾邮件。
假设黑名单的数量是数以亿计的,存放起来就是非常耗费存储空间的,布隆过滤器则是一个较好的解决方案。
把所有黑名单都放在布隆过滤器中,在收到邮件时,判断邮件地址是否在布隆过滤器中即可。
二、解决缓存穿透问题
把已存在数据的key存在布隆过滤器中,相当于redis前面挡着一个布隆过滤器。
当有新的请求时,先到布隆过滤器中查询是否存在:
如果布隆过滤器中不存在该条数据则直接返回;
如果布隆过滤器中已存在,才去查询缓存redis,如果redis里没查询到则再查询Mysql数据库
四、布隆过滤器优缺点
优点:高效插入和查询,内存占用空间少
缺点:
1、存在误判,不能精确过滤
2、不能删除元素
三、代码实现
一、手动实现布隆过滤器
/** * @Description : 布隆过滤器白名单初始化 * 1、初始化一部分数据进入到布隆过滤器 * 2、新增数据的时候如果数据库中没有,新增成功后,加入数据到布隆过滤器 * @Author : hc * @Date : 2023/6/14 22:00 **/ @Slf4j @Component public class BloomFilterInit { // 假设这是初始化数据 private static final String UID = "user:12"; // 白名单key public static final String WHITELIST_USER_KRY = "whitelist:user:"; @Resource private CheckUtils checkUtils; @Resource private RedisTemplate redisTemplate; /** * 白名单用户信息加载 * @author hc * @date 2023/6/15 11:45 */ @PostConstruct public void init() { // 1、获取hashCOde,由于可能出现负数,取绝对值 int abs = Math.abs(UID.hashCode()); // 2、直接设置布隆过滤器的bit数组为2的32次方,这里只使用一个hash函数与一个bit位置的数值进行演示 long index = checkUtils.getIndex(abs); // 3、使用redis新数据类型bitmap进行存储,key=WHITELIST_USER_KRY,偏移量表示这个bit数组的下标,value设置为true表示1 redisTemplate.opsForValue().setBit(WHITELIST_USER_KRY,index,Boolean.TRUE); } }
/** * @Description :布隆过滤器校验工具 * @Author : hc * @Date : 2023/6/15 11:34 **/ @Slf4j @Component public class CheckUtils { @Resource private RedisTemplate redisTemplate; /** * 布隆过滤器校验 * * @param key * @return boolean * @author hc * @date 2023/6/15 11:42 */ public boolean checkData(String key) { int abs = Math.abs(key.hashCode()); long index = (long) (abs % Math.pow(2, 32)); return redisTemplate.opsForValue().getBit(BloomFilterInit.WHITELIST_USER_KRY, index); } /** * 获取偏移量 * @param key * @return long * @author hc * @date 2023/6/15 17:19 */ public long getOffsetId(String key) { int abs = Math.abs(key.hashCode()); return getIndex(abs); } /** * 计算偏移量 * * @param abs * @return java.lang.Long * @author hc * @date 2023/6/15 16:25 */ public long getIndex(int abs) { if (0 == abs) { return 0L; } return (long) (abs % Math.pow(2, 32)); } }
/** * @Description : * @Author : hc * @Date : 2023/6/14 21:25 **/ @Slf4j @Service public class BloomFilterService { private static final String CACHE_KEY_USER = "user:"; @Resource private CheckUtils checkUtils; @Resource private RedisTemplate redisTemplate; @Resource private BloomFilterDao bloomFilterDao; @Transactional(rollbackFor = Exception.class) public void addUser(User user) { // 返回技术主键雪花id long i = bloomFilterDao.addUser(user); // 这里可以开启一个异步线程,在事务提交之后再进行操作 if (0 < i) { String key = CACHE_KEY_USER.concat(String.valueOf(user.getId())); long index = checkUtils.getOffsetId(key); // redis的数据都需要使用统一的json工具转成json格式后放入 String userJson = JSONUtil.toJsonStr(user); redisTemplate.opsForValue().set(key, userJson); redisTemplate.opsForValue().setBit(BloomFilterInit.WHITELIST_USER_KRY, index, Boolean.TRUE); log.info("新增用户信息|用户key:{}|布隆过滤器偏移量:{}", key, index); } } public User queryUser(Long id) { if (0 > id) { log.info("获取用户信息|用户id异常,异常id:{}", id); return null; } String key = CACHE_KEY_USER.concat(String.valueOf(id)); boolean checkData = checkUtils.checkData(key); if (!checkData) { log.info("获取用户信息|用户id不存在,异常id:{}", id); return null; } User user = (User) redisTemplate.opsForValue().get(key); if (Objects.isNull(user)) { // 这里可以换成分布式锁 synchronized (this) { user = (User) redisTemplate.opsForValue().get(key); if (Objects.isNull(user)) { user = bloomFilterDao.queryUser(id); } if (Objects.nonNull(user)) { long index = checkUtils.getOffsetId(key); String userJson = JSONUtil.toJsonStr(user); redisTemplate.opsForValue().set(key, userJson); redisTemplate.opsForValue().setBit(BloomFilterInit.WHITELIST_USER_KRY, index, Boolean.TRUE); } } } return user; } }
二、使用guava单机版实现布隆过滤器
1、误差率
/** * @Description : * @Author : hc * @Date : 2023/6/15 20:56 **/ @Service @Slf4j public class GuavaBloomFilterService { //布隆过滤器里预计要插入多少数据 public static int SIZE = 1000000; //误判率,它越小误判的个数也就越少(但是越小所消耗的资源就越多),这个数是谷歌布隆过滤器默认的值 //fpp the desired false positive probability public static double FPP = 0.03; public static void main(String[] args) { BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), SIZE, FPP); //1 先往布隆过滤器里面插入100万的样本数据 for (int i = 1; i <= SIZE; i++) { bloomFilter.put(i); } //故意取10万个不在过滤器里的值,看看有多少个会被认为在过滤器里 List<Integer> list = new ArrayList<>(SIZE); for (int i = SIZE + 1; i <= SIZE + (100000); i++) { if (bloomFilter.mightContain(i)) { log.info("被误判了:{}", i); list.add(i); } } log.info("误判的总数量::{}", list.size()); } } 误判率为:0.03BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), size,0.01); 误伤的数量:100
2、使用
/** * @Description : * @Author : hc * @Date : 2023/6/15 22:15 **/ public class GuavaBloomFilterUtils { private final static BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(UTF_8), 100000000, 0.01); public static boolean isExist(String id) { return bloomFilter.mightContain(id); } public static void put(String id) { bloomFilter.put(id); } }
/** * @Description : * @Author : hc * @Date : 2023/6/16 9:48 **/ @Slf4j @Configuration public class GuavaBloomFilterInterceptor implements HandlerInterceptor { @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { /** * 请求方式是OPTIONS说明是第一次,前端请求一次浏览器那边有两次请求, * 第一次请求方法携带OPTIONS,类似于先过来询问后端能不能连接,如果可以,则它会在HTTP头中包含一个名为“Allow”的头返回。 * 第二次请求才是get、post等真正的请求。 */ if (HttpMethod.OPTIONS.matches(request.getMethod())) { response.setStatus(HttpStatus.OK.value()); return Boolean.TRUE; } String dataStr = null; // 所有请求均使用post方式,if尽量不要嵌套进去 if (HttpMethod.POST.matches(request.getMethod())) { dataStr = request.getReader().lines().collect(Collectors.joining(System.lineSeparator())); } if (StrUtil.isEmpty(dataStr)) { resData(response, HttpStatus.CONTINUE.value()); return Boolean.FALSE; } // 假设是去其中的id JSONObject jsonObject = JSONUtil.parseObj(dataStr); String id = (String) jsonObject.get("id"); if (StrUtil.isNotEmpty(id) && GuavaBloomFilterUtils.isExist(id)) { return Boolean.TRUE; } resData(response,HttpStatus.CONTINUE.value()); return Boolean.FALSE; } /** * 统一使用HttpStatus.CONTINUE.value(),方便前端判断跳转指定页面 * @param response * @param status */ private static void resData(HttpServletResponse response, int status) { response.setStatus(status); response.setHeader("Content-Type", "application/json"); response.setCharacterEncoding("UTF-8"); log.info("布隆过滤器校验|数据不存在"); } }
/** * @Description : * @Author : hc * @Date : 2023/6/16 11:05 **/ @Configuration public class GuavaBloomFilterConfig implements WebMvcConfigurer { @Resource private GuavaBloomFilterInterceptor guavaBloomFilterInterceptor; @Override public void addInterceptors(InterceptorRegistry registry) { // 拦截所有请求,这里需要放行登录、注册等相关接口。 registry.addInterceptor(guavaBloomFilterInterceptor).addPathPatterns("/**"); } }
缺点:
1、基于本地缓存(jvm),容量受限制
2、多个应用就有多个布隆过滤器,多应用同步复杂。
三、redis分布式布隆过滤器的实现
二中的使用是基于jvm的,难以用到分布式系统中,如果想要应用于分布式系统中,就需要加入redis,使用redis的setBit命令即可对对应key设置bit位。也即是一、手动实现布隆过滤器中的实现。
可以参照guava版布隆过滤器源码
/** * @Description :思路:可以直接拿guava包里的源码进行修改 * 根据布隆过滤器原理 * 1、首先需要有k个函数,用来计算key对应的hash值,key与函数的关系是一对多 * 2、需要初始化一个N位的bit数组 * 3、新增key时,需要通过多个hash值对数组大小取余,找到对应多个位置,然后置为1 * 4、判断key是否在布隆过滤器中,用k个hash函数计算出k个散列值,并计算出对应的数组下表, * 查询数组中对应的数据,如果所有的比特位都是1,认为在集合中。 * @Author : hc * @Date : 2023/6/16 12:26 **/ public class BloomFilterHelper<T> { private Long bitSize; // 二进制数组大小 private int numHashFunctions; // hash函数个数 private Funnel<T> funnel; // 可自定义,如果只是String,Long等普通类型可以直接使用guava中的即可 /** * @param expectedInsertions 预估插入数据数量 * @param fpp 允许数据误差率 * @param funnel */ public BloomFilterHelper(Long expectedInsertions, double fpp, Funnel<T> funnel) { this.funnel = funnel; bitSize = optimalNumOfBits(expectedInsertions, fpp); numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, bitSize); } /** * 计算bit数组大小 * * @param n 预估插入数据数量 * @param p 允许数据误差率 * @return */ private long optimalNumOfBits(long n, double p) { if (p == 0) p = Double.MIN_VALUE; return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2))); } /** * 计算hash函数个数 * * @param n 预估插入数据数量 * @param m bit数组大小 * @return */ private int optimalNumOfHashFunctions(long n, long m) { return Math.max(1, (int) Math.round((double) m / n * Math.log(2))); } /** * 计算元素的hash散列下标 * * @param value 元素 * @return */ public Long[] mightContain(T value) { Long[] longs = new Long[numHashFunctions]; long hash64 = Hashing.murmur3_128().hashObject(value, funnel).asLong(); int hash1 = (int) hash64; int hash2 = (int) (hash64 >>> 32); // 循环hash函数,对数组取余,得到多个数组下标 for (int i = 1; i <= numHashFunctions; ++i) { int combinedHash = hash1 + i * hash2; if (combinedHash < 0) { combinedHash = ~combinedHash; } longs[i - 1] = combinedHash % bitSize; } return longs; } }
/** * @Description : * @Author : hc * @Date : 2023/6/16 16:10 **/ @Configuration public class RedisBloomFilterUtils { private static final Long SIZE = 100000000L; private static final Double FPP = 0.01; @Resource private RedisTemplate redisTemplate; private static final BloomFilterHelper bloomFilterHelper = new BloomFilterHelper(SIZE, FPP, Funnels.stringFunnel(Charsets.UTF_8)); /** * 布隆过滤器新增数据 * * @param key * @author hc * @date 2023/6/16 16:31 */ public void put(String key) { Long[] indexArray = getIndexArray(key); Arrays.stream(indexArray).filter(Objects::nonNull).forEach(index -> redisTemplate.opsForValue().setBit(key, index, Boolean.TRUE)); } /** * 检查布隆过滤器中是否存在 * * @param key * @return boolean * @author hc * @date 2023/6/16 16:42 */ public boolean mightContain(String key) { Long[] indexArray = getIndexArray(key); return !Arrays.stream(indexArray).filter(Objects::nonNull).anyMatch(index -> Boolean.FALSE == redisTemplate.opsForValue().getBit(key, index)); } private static Long[] getIndexArray(String key) { Assert.isFalse(StrUtil.isEmpty(key), "布隆过滤器新增数据|key为空"); // 获取数组下标 return bloomFilterHelper.mightContain(key); } }
/** * @Description : 初始化数据 * @Author : hc * @Date : 2023/6/16 17:28 **/ @Slf4j @Configuration public class RedisBloomFilterInit implements InitializingBean { private static final String PRE_KEY = "user:"; @Resource private RedisBloomFilterUtils redisBloomFilterUtils; @Override public void afterPropertiesSet() throws Exception { List<String> list = Lists.newArrayList("1", "2"); log.info("加载数据到布隆过滤器,size:{}", list.size()); list.stream().filter(Objects::nonNull).forEach(id -> { String key = PRE_KEY.concat(id); redisBloomFilterUtils.put(key); }); } }
/** * @Description : * @Author : hc * @Date : 2023/6/16 9:48 **/ @Slf4j @Configuration public class RedisBloomFilterInterceptor implements HandlerInterceptor { private static final String PRE_KEY = "user:"; @Resource private RedisBloomFilterUtils redisBloomFilterUtils; @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { /** * 请求方式是OPTIONS说明是第一次,前端请求一次浏览器那边有两次请求, * 第一次请求方法携带OPTIONS,类似于先过来询问后端能不能连接,如果可以,则它会在HTTP头中包含一个名为“Allow”的头返回。 * 第二次请求才是get、post等真正的请求。 */ if (HttpMethod.OPTIONS.matches(request.getMethod())) { response.setStatus(HttpStatus.OK.value()); return Boolean.TRUE; } String dataStr = null; // 所有请求均使用post方式,if尽量不要嵌套进去 if (HttpMethod.POST.matches(request.getMethod())) { dataStr = request.getReader().lines().collect(Collectors.joining(System.lineSeparator())); } if (StrUtil.isEmpty(dataStr)) { resData(response, HttpStatus.CONTINUE.value()); return Boolean.FALSE; } // 假设是去其中的id JSONObject jsonObject = JSONUtil.parseObj(dataStr); String id = (String) jsonObject.get("id"); String key = PRE_KEY.concat(id); if (StrUtil.isNotEmpty(id) && redisBloomFilterUtils.mightContain(key)) { return Boolean.TRUE; } resData(response,HttpStatus.CONTINUE.value()); return Boolean.FALSE; } /** * 统一使用HttpStatus.CONTINUE.value(),方便前端判断跳转指定页面 * @param response * @param status */ private static void resData(HttpServletResponse response, int status) { response.setStatus(status); response.setHeader("Content-Type", "application/json"); response.setCharacterEncoding("UTF-8"); log.info("布隆过滤器校验|数据不存在"); } }
/** * @Description : * @Author : hc * @Date : 2023/6/16 11:05 **/ @Configuration public class RedisBloomFilterConfig implements WebMvcConfigurer { @Resource private RedisBloomFilterInterceptor redisBloomFilterInterceptor; @Override public void addInterceptors(InterceptorRegistry registry) { // 拦截所有请求,这里需要放行登录、注册等相关接口。 registry.addInterceptor(redisBloomFilterInterceptor).addPathPatterns("/**"); } }
四、利用Redisson实现分布式布隆过滤器
/** * @Description : * @Author : hc * @Date : 2023/6/24 21:20 **/ @Configuration public class RedissonBloomFilterInit { private static final String BLOOM_FILTER_KEY = "bloom_filter_key:"; @Resource private Redisson redisson; @ConditionalOnBean(name = "redisson") @Bean("bloomFilter") public RBloomFilter<Object> initBloomFilter() { RBloomFilter<Object> bloomFilter = getBloomFilter(BLOOM_FILTER_KEY); bloomFilter.tryInit(100000000L, 0.01); return bloomFilter; } private RBloomFilter<Object> getBloomFilter(String key) { return redisson.getBloomFilter(key); } }
/** * @Description : * @Author : hc * @Date : 2023/6/24 21:00 **/ @Slf4j @Component public class RedissonBloomFilterUtils { @Resource private RBloomFilter<Object> bloomFilter; public boolean put(String value) { if (StringUtil.isEmpty(value)) { log.warn("分布式布隆过滤器|value不能为空"); return false; } return bloomFilter.add(value); } public boolean contain(String value) { if (StringUtil.isEmpty(value)) { log.warn("分布式布隆过滤器|value不能为空"); return false; } return bloomFilter.contains(value); } }
四、总结
一、 对布隆过滤器的的总结
无论使用谷歌版本的布隆过滤器还是自己编写的,都会存在两个问题,
1、因为不同元素经过hash函数计算后可能会出现相同的hash值(hash碰撞),就会出现一个误判率的问题
2、因为有hash碰撞,导致同一个位置可能存放不同的数据,这对于删除操作是很不友好的。
对于这些情况可查看另一种布隆过滤器,布谷鸟过滤器
二、由布隆过滤器延伸的思考
1、如果在项目中初始化一个布隆过滤器,假设大小为10000000,当项目中数据一直在新增,一直布隆过滤器中put值,总有一天hash碰撞的概率会提高,误判率也就随之提高。但是在大数据量的布隆过滤器,进行删除重建,这成本无疑是很高的。
2、对于缓存击穿,如果在使用srpingcache的注解后,可以在主键中配置单个线程访问mysql。
@Cacheable(cacheNames="menu",sync="true")3、对于使用了springcache注解的,想要解决缓存穿透问题可以,设置返回值为null,
spring.cache.redis.cache-null-values=true然后时间设置短一点,但是这有个问题就是:如果在设置的null失效的时间内,有大量的请求进来,且查出来的数据也都是null,这个时候,redis内存也会飙升。在这个情况下可以考虑一下布隆过滤器的使用了,加一个黑名单操作。