java布隆过滤器解决redis缓存穿透,spring boot 布隆过滤器
00-1010什么是恶意流量渗透，如何防范Bloom Filter？另一个使用Bloom filter到Redis的场景，使用Redis中的Bloom Filter和SpringBoot构建spring boot项目，使用压力测量工具将120万条数据馈入Redis Bloomfilter看实际效果。本文主要介绍SpringBoot Redis Bloom filter防止恶意流量突破缓存，具体如下：3360
00-1010假设我们的Redis中有一组用户的注册邮件，以email为关键字，它对应着DB中user表的一些字段。
一般来说，当一个合理的请求过来时，我们会先判断用户是否是Redis中的会员，因为从缓存中读取数据速度很快。如果这个成员在缓存中不存在，那么我们将在DB中检查它。
现在，想象一下，有成千上万个不同IP的请求(不要以为一个都没有，我们在2018年和2019年遇到过，因为攻击的成本很低)用一个根本不存在于Redis的密钥访问你的网站。这时，让我们想象一下：
请求到达Web服务器；将请求发送给应用层-微服务层；请求从Redis获取数据。这把钥匙；Redis中不存在；于是请求到达DB层，在DB中建立连接后，会进行几千万甚至上亿个DB连接请求的查询。不管Redis能不能撑得住，DB都会瞬间爆炸。这就是“Redis穿透”，又称“缓存击穿”。会炸掉你的缓存甚至DB，造成一系列的“雪崩效应”。
00-1010即使用bloom filter，可以将用户表中的所有关键查询字段放入Redis的Bloom filter中。有人会说，这不是疯了吗，我有4000万会员？那又怎样！
你把4000会员放在Redis里，这是相当夸张的。有些网站有8000万或1亿会员。所以我没有让你直接放在Redis，而是放在Bloom filter！
在Bloom filter中，键和值不是直接放进去的，但是Bloom filter中存储的内容是这样的：
BloomFilter是一种空间有效的概率数据结构，由Burton Howard Bloom于1970年提出。通常用于确定一个元素是否在集合中。它有很高的空间效率，但会带来假阳性错误。
False positiveFalse negatives布鲁姆菲特为了节省空间牺牲了一定的准确性。所以就带来了假阳性的问题。
当False positive布鲁姆滤波判断一个元素在一个集合中时，会有一定的错误率，称为假阳性。通常缩写为fpp。
False negativesBloomFilter确定元素不在集合中时的错误率。BloomFilter确定该元素不在集合中，因此该元素一定不在集合中。所以，负的概率是0。
BloomFilter使用长度为m bit的字节数组，使用k个哈希函数，添加一个元素：通过k个哈希将元素映射到字节数组中的k个位置，并将对应位置的字节设置为1。查询元素是否存在：哈希元素k次得到k个位置。如果对应于k个位置的位是1，则认为它存在；否则视为不存在。
因为所有的位都存储在里面，所以数据量会很小。到什么程度？写这篇博客的时候，我把100万封邮件插入Redis的bloom filter，只用了不到3Mb。
布隆过滤器将有几个关键值。根据这个值，你可以大致算出，当它的误伤率为时，会放多少条数据，会占用多少系统资源。这个算法有个网站：https://krisives.github.io/bloom-calculator/.我们放入100万条数据，假设意外伤害率为0.001%。看，它自动获取Redis需要申请的系统内存资源量。
那么如何解决这个意外伤害率呢？很简单，当出现意外伤害时，业务或运营会报告意外伤害率。这时候你只需要添加一个白名单就可以了。相对于100万条数据，1000份白名单不是问题。并且bloom filter的返回速度是over block，也就是说在80-100毫秒内，调用端的Key存在或不存在。
目录
假设我用python爬虫爬了4亿个网址，需要复制吗？
看，这个场景用的是布鲁姆滤镜。
让我们开始Redis BloomFilter的旅程。
什么是恶意流量穿透
Redis从4.0开始只支持bloom fil。
ter，因此本例中我们使用的是Redis5.4。
Redis的bloom filter下载地址在这：https://github.com/RedisLabsModules/redisbloom.git
git clone https://github.com/RedisLabsModules/redisbloom.gitcd redisbloommake # 编译
让Redis启动时可以加载bloom filter有两种方式：
手工加载式：
redis-server --loadmodule ./redisbloom/rebloom.so
每次启动自加载：
编辑Redis的redis.conf文件，加入：
loadmodule /soft/redisbloom/redisbloom.so
Like this:
在Redis里使用Bloom Filter
基本指令：
bf.reserve {key} {error_rate} {size}
127.0.0.1:6379> bf.reserve userid 0.01 100000OK
上面这条命令就是：创建一个空的布隆过滤器，并设置一个期望的错误率和初始大小。{error_rate}过滤器的错误率在0-1之间，如果要设置0.1%，则应该是0.001。该数值越接近0，内存消耗越大，对cpu利用率越高。
bf.add {key} {item}
127.0.0.1:6379> bf.add userid 181920(integer) 1
上面这条命令就是：往过滤器中添加元素。如果key不存在，过滤器会自动创建。
bf.exists {key} {item}
127.0.0.1:6379> bf.exists userid 101310299(integer) 1
上面这条命令就是：判断指定key的value是否在bloomfilter里存在。存在：返回1，不存在：返回0。
结合SpringBoot使用
网上很多写的都是要么是直接使用jedis来操作的，或者是java里execute一个外部进程来调用Redis的bloom filter指令的。很多都是调不通或者helloworld一个级别的，是根本无法上生产级别应用的。
笔者给出的代码保障读者完全可用！
笔者不是数学家，因此就借用了google的guava包来实现了核心算法，核心代码如下：
BloomFilterHelper.java
package org.sky.platform.util; import com.google.common.base.Preconditions;import com.google.common.hash.Funnel;import com.google.common.hash.Hashing; public class BloomFilterHelper<T> {private int numHashFunctions; private int bitSize; private Funnel<T> funnel; public BloomFilterHelper(Funnel<T> funnel, int expectedInsertions, double fpp) {Preconditions.checkArgument(funnel != null, "funnel不能为空");this.funnel = funnel;bitSize = optimalNumOfBits(expectedInsertions, fpp);numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, bitSize);} int[] murmurHashOffset(T value) {int[] offset = new int[numHashFunctions]; long hash64 = Hashing.murmur3_128().hashObject(value, funnel).asLong();int hash1 = (int) hash64;int hash2 = (int) (hash64 >>> 32);for (int i = 1; i <= numHashFunctions; i++) {int nextHash = hash1 + i * hash2;if (nextHash < 0) {nextHash = ~nextHash;}offset[i - 1] = nextHash % bitSize;} return offset;} /** * 计算bit数组的长度 */private int optimalNumOfBits(long n, double p) {if (p == 0) {p = Double.MIN_VALUE;}return (int) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));} /** * 计算hash方法执行次数 */private int optimalNumOfHashFunctions(long n, long m) {return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));}}下面放出全工程解说，我已经将源码上传到了我的git上了，确保读者可用，源码地址在这：https://github.com/mkyuangithub/mkyuangithub.git
搭建spring boot工程
项目Redis配置
我们在redis-practice工程里建立一个application.properties文件，内容如下：
spring.redis.database=0 spring.redis.host=192.168.56.101spring.redis.port=6379spring.redis.password=111111spring.redis.pool.max-active=10 spring.redis.pool.max-wait=-1 spring.redis.pool.max-idle=10 spring.redis.pool.min-idle=0 spring.redis.timeout=1000
以上这个是demo环境的配置。
我们此处依旧使用的是在前一篇springboot+nacos+dubbo实现异常统一管理中的xxx-project->sky-common->nacos-parent的依赖结构。
在redis-practice工程的org.sky.config包中放入redis的springboot配置
RedisConfig.java
package org.sky.config; import com.fasterxml.jackson.annotation.JsonAutoDetect;import com.fasterxml.jackson.annotation.PropertyAccessor;import com.fasterxml.jackson.databind.ObjectMapper;import org.springframework.cache.CacheManager;import org.springframework.cache.annotation.CachingConfigurerSupport;import org.springframework.cache.annotation.EnableCaching;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.data.redis.cache.RedisCacheManager;import org.springframework.data.redis.connection.RedisConnectionFactory;import org.springframework.data.redis.core.*;import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;import org.springframework.data.redis.serializer.StringRedisSerializer; @Configuration@EnableCachingpublic class RedisConfig extends CachingConfigurerSupport {/** * 选择redis作为默认缓存工具 * * @param redisTemplate * @return */@Beanpublic CacheManager cacheManager(RedisTemplate redisTemplate) {RedisCacheManager rcm = new RedisCacheManager(redisTemplate);return rcm;} /** * retemplate相关配置 * * @param factory * @return */@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) { RedisTemplate<String, Object> template = new RedisTemplate<>();// 配置连接工厂template.setConnectionFactory(factory); // 使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值（默认使用JDK的序列化方式）Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class); ObjectMapper om = new ObjectMapper();// 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和publicom.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);// 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会跑出异常om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jacksonSeial.setObjectMapper(om); // 值采用json序列化template.setValueSerializer(jacksonSeial);// 使用StringRedisSerializer来序列化和反序列化redis的key值template.setKeySerializer(new StringRedisSerializer()); // 设置hash key 和value序列化模式template.setHashKeySerializer(new StringRedisSerializer());template.setHashValueSerializer(jacksonSeial);template.afterPropertiesSet(); return template;} /** * 对hash类型的数据操作 * * @param redisTemplate * @return */@Beanpublic HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForHash();} /** * 对redis字符串类型数据操作 * * @param redisTemplate * @return */@Beanpublic ValueOperations<String, Object> valueOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForValue();} /** * 对链表类型的数据操作 * * @param redisTemplate * @return */@Beanpublic ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForList();} /** * 对无序集合类型的数据操作 * * @param redisTemplate * @return */@Beanpublic SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForSet();} /** * 对有序集合类型的数据操作 * * @param redisTemplate * @return */@Beanpublic ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForZSet();}}这个配置除实现了springboot自动发现redis在application.properties中的配置外我们还添加了不少redis基本的数据结构的操作的封装。
我们为此还要再封装一套Redis Util小组件，它们位于sky-common工程中
RedisUtil.java
package org.sky.platform.util; import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.stereotype.Component; import java.util.Collection;import java.util.Date;import java.util.Set;import java.util.concurrent.TimeUnit;import java.util.stream.Collectors;import java.util.stream.Stream;import com.google.common.base.Preconditions;import org.springframework.data.redis.core.RedisTemplate; @Componentpublic class RedisUtil {@Autowiredprivate RedisTemplate<String, String> redisTemplate; /** * 默认过期时长，单位：秒 */public static final long DEFAULT_EXPIRE = 60 * 60 * 24; /** * 不设置过期时长 */public static final long NOT_EXPIRE = -1; public boolean existsKey(String key) {return redisTemplate.hasKey(key);} /** * 重名名key，如果newKey已经存在，则newKey的原值被覆盖 * * @param oldKey * @param newKey */public void renameKey(String oldKey, String newKey) {redisTemplate.rename(oldKey, newKey);} /** * newKey不存在时才重命名 * * @param oldKey * @param newKey * @return 修改成功返回true */public boolean renameKeyNotExist(String oldKey, String newKey) {return redisTemplate.renameIfAbsent(oldKey, newKey);} /** * 删除key * * @param key */public void deleteKey(String key) {redisTemplate.delete(key);} /** * 删除多个key * * @param keys */public void deleteKey(String... keys) {Set<String> kSet = Stream.of(keys).map(k -> k).collect(Collectors.toSet());redisTemplate.delete(kSet);} /** * 删除Key的集合 * * @param keys */public void deleteKey(Collection<String> keys) {Set<String> kSet = keys.stream().map(k -> k).collect(Collectors.toSet());redisTemplate.delete(kSet);} /** * 设置key的生命周期 * * @param key * @param time * @param timeUnit */public void expireKey(String key, long time, TimeUnit timeUnit) {redisTemplate.expire(key, time, timeUnit);} /** * 指定key在指定的日期过期 * * @param key * @param date */public void expireKeyAt(String key, Date date) {redisTemplate.expireAt(key, date);} /** * 查询key的生命周期 * * @param key * @param timeUnit * @return */public long getKeyExpire(String key, TimeUnit timeUnit) {return redisTemplate.getExpire(key, timeUnit);} /** * 将key设置为永久有效 * * @param key */public void persistKey(String key) {redisTemplate.persist(key);} /** * 根据给定的布隆过滤器添加值 */public <T> void addByBloomFilter(BloomFilterHelper<T> bloomFilterHelper, String key, T value) {Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");int[] offset = bloomFilterHelper.murmurHashOffset(value);for (int i : offset) {redisTemplate.opsForValue().setBit(key, i, true);}} /** * 根据给定的布隆过滤器判断值是否存在 */public <T> boolean includeByBloomFilter(BloomFilterHelper<T> bloomFilterHelper, String key, T value) {Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");int[] offset = bloomFilterHelper.murmurHashOffset(value);for (int i : offset) {if (!redisTemplate.opsForValue().getBit(key, i)) {return false;}} return true;}}RedisKeyUtil.java
package org.sky.platform.util; public class RedisKeyUtil {/** * redis的key 形式为： 表名:主键名:主键值:列名 * * @param tableName 表名 * @param majorKey 主键名 * @param majorKeyValue 主键值 * @param column 列名 * @return */public static String getKeyWithColumn(String tableName, String majorKey, String majorKeyValue, String column) {StringBuffer buffer = new StringBuffer();buffer.append(tableName).append(":");buffer.append(majorKey).append(":");buffer.append(majorKeyValue).append(":");buffer.append(column);return buffer.toString();} /** * redis的key 形式为： 表名:主键名:主键值 * * @param tableName 表名 * @param majorKey 主键名 * @param majorKeyValue 主键值 * @return */public static String getKey(String tableName, String majorKey, String majorKeyValue) {StringBuffer buffer = new StringBuffer();buffer.append(tableName).append(":");buffer.append(majorKey).append(":");buffer.append(majorKeyValue).append(":");return buffer.toString();}}然后就是制作 redis里如何使用BloomFilter的BloomFilterHelper.java了，它也位于sky-common文件夹，源码如上已经贴了，因此此处就不再作重复。
最后我们在sky-common里放置一个UserVO用于演示
UserVO.java
package org.sky.vo; import java.io.Serializable; public class UserVO implements Serializable { private String name;private String address;private Integer age;private String email = ""; public String getEmail() {return email;} public void setEmail(String email) {this.email = email;} public String getName() {return name;} public void setName(String name) {this.name = name;} public String getAddress() {return address;} public void setAddress(String address) {this.address = address;} public Integer getAge() {return age;} public void setAge(Integer age) {this.age = age;} }下面给出我们所有gitrepo里依赖的nacos-parent的pom.xml文件内容，此次我们增加了对于spring-boot-starter-data-redis，它跟着我们的全局springboot版本走：
parent工程的pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.sky.demo</groupId><artifactId>nacos-parent</artifactId><version>0.0.1-SNAPSHOT</version><packaging>pom</packaging><description>Demo project for Spring Boot Dubbo Nacos</description><modules></modules> <properties><java.version>1.8</java.version><spring-boot.version>1.5.15.RELEASE</spring-boot.version><dubbo.version>2.7.3</dubbo.version><curator-framework.version>4.0.1</curator-framework.version><curator-recipes.version>2.8.0</curator-recipes.version><druid.version>1.1.20</druid.version><guava.version>27.0.1-jre</guava.version><fastjson.version>1.2.59</fastjson.version><dubbo-registry-nacos.version>2.7.3</dubbo-registry-nacos.version><nacos-client.version>1.1.4</nacos-client.version><mysql-connector-java.version>5.1.46</mysql-connector-java.version><disruptor.version>3.4.2</disruptor.version><aspectj.version>1.8.13</aspectj.version><nacos-service.version>0.0.1-SNAPSHOT</nacos-service.version><spring.data.redis>1.8.14-RELEASE</spring.data.redis><skycommon.version>0.0.1-SNAPSHOT</skycommon.version><maven.compiler.source>${java.version}</maven.compiler.source><maven.compiler.target>${java.version}</maven.compiler.target><compiler.plugin.version>3.8.1</compiler.plugin.version><war.plugin.version>3.2.3</war.plugin.version><jar.plugin.version>3.1.2</jar.plugin.version><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding></properties><dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>${spring-boot.version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>${spring-boot.version}</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>org.apache.dubbo</groupId><artifactId>dubbo-spring-boot-starter</artifactId><version>${dubbo.version}</version><exclusions><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.dubbo</groupId><artifactId>dubbo</artifactId><version>${dubbo.version}</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>${curator-framework.version}</version></dependency> <dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>${curator-recipes.version}</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>${mysql-connector-java.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>${druid.version}</version></dependency><dependency><groupId>com.lmax</groupId><artifactId>disruptor</artifactId><version>${disruptor.version}</version></dependency><dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>${guava.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><dependency><groupId>org.apache.dubbo</groupId><artifactId>dubbo-registry-nacos</artifactId><version>${dubbo-registry-nacos.version}</version></dependency><dependency><groupId>com.alibaba.nacos</groupId><artifactId>nacos-client</artifactId><version>${nacos-client.version}</version></dependency><dependency><groupId>org.aspectj</groupId><artifactId>aspectjweaver</artifactId><version>${aspectj.version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>${spring-boot.version}</version></dependency></dependencies></dependencyManagement><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>${compiler.plugin.version}</version><configuration><source>${java.version}</source><target>${java.version}</target></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-war-plugin</artifactId><version>${war.plugin.version}</version></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-jar-plugin</artifactId><version>${jar.plugin.version}</version></plugin></plugins></build></project>sky-common中pom.xml文件
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.sky.demo</groupId><artifactId>skycommon</artifactId><version>0.0.1-SNAPSHOT</version><parent><groupId>org.sky.demo</groupId><artifactId>nacos-parent</artifactId><version>0.0.1-SNAPSHOT</version></parent><dependencies> <dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId></dependen
郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。