一、分布式锁
在并发编程中，通过锁，来避免由于竞争而造成的数据不一致问题
1.1、高并发下单超卖问题
@Autowired
RedisTemplate<String, String> redisTemplate;
String key = "maotai20210319001";//茅台商品编号
ScheduledExecutorService executorService;
ConcurrentSkipListSet<String> set = new ConcurrentSkipListSet<>();
@PostConstruct
void init() {
redisTemplate.opsForValue().set(key, "100");
executorService = Executors.newScheduledThreadPool(1);
String renewlua = "" +
"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('expire',KEYS[1],ARGV[2]) ; return true " +
"else return false " +
"end";
executorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
Iterator<String> iterator = set.iterator();
while (iterator.hasNext()) {
String clientid = iterator.next();
Boolean renew = redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
Boolean eval = null;
try {
eval = redisConnection.eval(
renewlua.getBytes(),
ReturnType.BOOLEAN,
1,
lockKey.getBytes(),
clientid.getBytes(),
"5".getBytes()
);
} catch (Exception e) {
e.printStackTrace();
}
return eval;
}
});
}
}
}, 0, 1, TimeUnit.SECONDS);
}
问题分析
-
现象：本地锁在多节点下失效(集群/分布式)
-
原因：本地锁它只能锁住本地JVM进程中的多个线程，对于多个JVM进程的不同线程间是锁不住的
-
解决：分布式锁（在分布式环境下提供锁服务，并且达到本地锁的效果）
1.2、何为分布式锁
-
当在分布式架构下，数据只有一份（或有限制），此时需要利用锁的技术控制某一时刻修改数据的进程数。
-
用一个状态值表示锁，对锁的占用和释放通过状态值来标识。
1.3、分布式锁特点
-
互斥性：不仅要在同一jvm进程下的不同线程间互斥，更要在不同jvm进程下的不同线程间互斥。
-
锁超时：支持锁的自动释放，防止死锁。
-
正确，高效，高可用：解铃还须系铃人（加锁和解锁必须是同一个线程），加锁和解锁操作一定要高效，提供锁的服务要具备容错性。
-
可重入：如果一个线程拿到了锁之后继续去获取锁还能获取到，我们称锁是可重入的（方法的递归调用）。
-
阻塞/非阻塞：如果获取不到直接返回视为非阻塞的，如果获取不到会等待锁的释放直到获取锁或者等待超时，视为阻塞的。
-
公平/非公平：按照请求的顺序获取锁视为公平的。
1.4、基于Redis实现分布式锁
1.4.1、实现思路：
锁的实现主要基于redis的 SETNX 命令
SETNX key value
将 key 的值设为 value ，当且仅当 key 不存在。
若给定的 key 已经存在，则 SETNX 不做任何动作。
SETNX 是『SET if Not eXists』(如果不存在，则 SET)的简写。
返回值：
设置成功，返回 1
设置失败，返回 0
使用 SETNX 完成同步锁的流程及事项如下
-
使用 SETNX 命令获取锁，若返回0（key已存在，锁已存在）则获取失败，反之获取成功。
-
为了防止获取锁后程序出现异常，导致其他线程/进程调用 SETNX 命令总是返回0而进入死锁状态，需要为该key设置一个“合理”的过期时间。
-
释放锁，使用 DEL 命令将锁数据删除。
1.4.2、实现代码版本
@GetMapping("/get/maotai3")
public String seckillMaotai3() {
//先获取锁,如果能获取到则进行业务操作
// Boolean execute = redisTemplate.execute(new RedisCallback<Boolean>() {
// @Override
// public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
// Boolean aBoolean = redisConnection.setNX(lockKey.getBytes(StandardCharsets.UTF_8), "1".getBytes(StandardCharsets.UTF_8));
// redisConnection.expire(lockKey.getBytes(StandardCharsets.UTF_8),10);
// return aBoolean;
// }
// });
String clientId = UUID.randomUUID().toString() + Thread.currentThread().getId();
Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 5, TimeUnit.SECONDS);
if (ifAbsent) {
//获取到了锁,给锁设置一个过期时间
// redisTemplate.expire(lockKey,10,TimeUnit.SECONDS);
//开始进行业务操作
try {
Integer count = Integer.parseInt(redisTemplate.opsForValue().get(key));
//如果还有库存
if (count > 0) {
//抢到了茅台，库存减一
redisTemplate.opsForValue().set(key, String.valueOf(count - 1));
//后续操作 do something
log.info("我抢到茅台了!");
//模拟故障退出
System.exit(1);
return "ok";
} else {
return "no";
}
} catch (NumberFormatException e) {
e.printStackTrace();
} finally {
//保证锁一定会被释放
// 自己的锁自己释放 但是需要保证原子操作，此处无法保证
String lockvalue = redisTemplate.opsForValue().get(lockKey);
if (lockvalue != null && lockvalue.equals(clientId)) {
redisTemplate.delete(lockKey);
}
}
}
return "dont get lock";
}
问题分析
-
setnx 和 expire是非原子性操作（解决：2.6以前可用使用lua脚本，2.6以后可用set命令）。
-
错误解锁（如何保证解铃还须系铃人：给锁加一个唯一标识）。
1.4.3、错误解锁问题解决
@GetMapping("/get/maotai4")
public String seckillMaotai4() {
//先获取锁,如果能获取到则进行业务操作
String clientId = UUID.randomUUID().toString() + Thread.currentThread().getId();
/*String locklua = "" +
"if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then redis.call('expire',KEYS[1],ARGV[2]) ; return true " +
"else return false " +
"end";
Boolean islock = redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
Boolean lock = redisConnection.eval(
locklua.getBytes(),
ReturnType.BOOLEAN,
1,
lockKey.getBytes(),
clientId.getBytes(),
"5".getBytes()
);
return lock;
}
});*/
Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 5, TimeUnit.SECONDS);//可以改用lua脚本，不改也可以
if (islock) {
//开始进行业务操作
try {
Integer count = Integer.parseInt(redisTemplate.opsForValue().get(key));
//如果还有库存
if (count > 0) {
//抢到了茅台，库存减一
redisTemplate.opsForValue().set(key, String.valueOf(count - 1));
//后续操作 do something
log.info("我抢到茅台了!");
//模拟故障退出
// System.exit(1);
return "ok";
} else {
return "no";
}
} catch (NumberFormatException e) {
e.printStackTrace();
} finally {
//保证锁一定会被释放
String unlocklua = "" +
"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('del',KEYS[1]); return true " +
"else return false " +
"end";
Boolean unlock = redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
Boolean unlock = redisConnection.eval(unlocklua.getBytes(),
ReturnType.BOOLEAN,
1,
lockKey.getBytes(),
clientId.getBytes());
return unlock;
}
});
}
}
return "dont get lock";
}
1.4.4、锁续期/锁续命
/**
* 3,锁续期/锁续命
* 拿到锁之后执行业务，业务的执行时间超过了锁的过期时间
*
* 如何做？
* 给拿到锁的线程创建一个守护线程(看门狗)，守护线程定时/延迟 判断拿到锁的线程是否还继续持有锁，如果持有则为其续期
*
*/
//模拟一下守护线程为其续期
ScheduledExecutorService executorService;//创建守护线程池
ConcurrentSkipListSet<String> set = new ConcurrentSkipListSet<String>();//队列
@PostConstruct
public void init2(){
executorService = Executors.newScheduledThreadPool(1);
//编写续期的lua
String expirrenew = "" +
"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('expire',KEYS[1],ARGV[2]) ; return true " +
"else return false " +
"end";
executorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
Iterator<String> iterator = set.iterator();
while (iterator.hasNext()) {
String rquestid = iterator.next();
redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
Boolean eval = false;
try {
eval = redisConnection.eval(
expirrenew.getBytes(),
ReturnType.BOOLEAN,
1,
lockey.getBytes(),
rquestid.getBytes(),
"5".getBytes()
);
} catch (Exception e) {
log.error("锁续期失败,{}",e.getMessage());
}
return eval;
}
});
}
}
},0,1,TimeUnit.SECONDS);
}
@GetMapping("/get/maotai5")
public String seckillMaotai5() {
String requestid = UUID.randomUUID().toString() + Thread.currentThread().getId();
//获取锁
Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockey,requestid,5,TimeUnit.SECONDS);
if (islock) {
//获取锁成功后让守护线程为其续期
set.add(requestid);
try {
Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1
//如果还有库存
if (count > 0) {
//抢到了茅台，库存减一
redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));
//后续操作 do something
//seckillMaotai5();
//模拟业务超时
TimeUnit.SECONDS.sleep(10);
log.info("我抢到茅台了!");
return "ok";
}else {
return "no";
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//解除锁续期
set.remove(requestid);
//释放锁
String unlocklua = "" +
"if redis.call('get',KEYS[1]) == ARGV[1] then redis.call('del',KEYS[1]) ; return true " +
"else return false " +
"end";
redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
Boolean eval = redisConnection.eval(
unlocklua.getBytes(),
ReturnType.BOOLEAN,
1,
lockey.getBytes(),
requestid.getBytes()
);
return eval;
}
});
}
}
return "dont get lock";
}
1.4.5、锁的可重入/阻塞锁（redisson）
/**
*
* 4,如何支持可重入
* 重入次数/过期时间
* 获取
* 获取
* 获取
*
* 释放
* 释放
* 释放
*
* 基于本地实现
* 还是基于redis但是更换了数据类型，采用hash类型来实现
* key field value
* 锁key 请求id 重入次数
* 用lua实现
*
*
* 5,阻塞/非阻塞的问题：现在的锁是非阻塞的，一旦获取不到锁直接返回了
* 如何做一个阻塞锁呢？
* 获取不到就等待锁的释放，直到获取到锁或者等待超时
* 1：基于客户端轮询的方案
* 2：基于redis的发布/订阅方案
*
*
* 有没有好的实现呢？
* Redisson
*
*/
@Value("${spring.redis.host}")
String host;
@Value("${spring.redis.port}")
String port;
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer().setAddress("redis://"+host+":"+port);
return Redisson.create(config);
}
@Autowired
RedissonClient redissonClient;
@GetMapping("/get/maotai6")
public String seckillMaotai6() {
//要去获取锁
RLock lock = redissonClient.getLock(lockey);
lock.lock();
try {
Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai)); // 1
//如果还有库存
if (count > 0) {
//抢到了茅台，库存减一
redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));
//后续操作 do something
log.info("我抢到茅台了!");
return "ok";
}else {
return "no";
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();;
}
return "";
}
1.5、redisson
Redisson内置了一系列的 分布式对象 ， 分布式集合 ， 分布式锁 ， 分布式服务
等诸多功能特性，是一款基于Redis实现，拥有一系列分布式系统功能特性的工具包，是实现分布式系统架构中缓存中间件的最佳选择。
下载地址：https://github.com/redisson/redisson
加解锁源码解析参考: 传送门
实现
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.8.2</version>
</dependency>
@Value("${spring.redis.host}")
String host;
@Value("${spring.redis.port}")
String port;
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer().setAddress("redis://"+host+":"+port);
return Redisson.create(config);
}
@Autowired
RedissonClient redissonClient;
@GetMapping("/get/maotai6")
public String seckillMaotai6() {
RLock lock = redissonClient.getLock(lockey);
//获取锁
lock.lock();
try {
Integer count = Integer.parseInt(redisTemplate.opsForValue().get(maotai));
//如果还有库存
if (count > 0) {
//抢到了茅台，库存减一
redisTemplate.opsForValue().set(maotai,String.valueOf(count-1));
//后续操作 do something
log.info("我抢到茅台了!");
return "ok";
}else {
return "no";
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return "dont get lock";
}
源码剖析
- 加锁的（是否支持重入）
- 锁续期的
- 阻塞获取
- 释放
/**
* 原理
* 1，加锁
* <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
* internalLockLeaseTime = unit.toMillis(leaseTime);
*
* return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
* #如果锁key不存在
* "if (redis.call('exists', KEYS[1]) == 0) then " +
* #设置锁key,field是唯一标识,value是重入次数
* "redis.call('hset', KEYS[1], ARGV[2], 1); " +
* #设置锁key的过期时间 默认30s
* "redis.call('pexpire', KEYS[1], ARGV[1]); " +
* "return nil; " +
* "end; " +
* #如果锁key存在
* "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
* #重入次数+1
* "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
* #重置过期时间
* "redis.call('pexpire', KEYS[1], ARGV[1]); " +
* "return nil; " +
* "end; " +
* "return redis.call('pttl', KEYS[1]);",
* Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
* }
*
* 2，锁续期
* private void scheduleExpirationRenewal(final long threadId) {
* if (expirationRenewalMap.containsKey(getEntryName())) {
* return;
* }
*
* Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
* @Override
* public void run(Timeout timeout) throws Exception {
* //续期函数的真正实现
* RFuture<Boolean> future = renewExpirationAsync(threadId);
*
* future.addListener(new FutureListener<Boolean>() {
* @Override
* public void operationComplete(Future<Boolean> future) throws Exception {
* expirationRenewalMap.remove(getEntryName());
* if (!future.isSuccess()) {
* log.error("Can't update lock " + getName() + " expiration", future.cause());
* return;
* }
*
* if (future.getNow()) {
* // reschedule itself 再次调用自己,最终形成的结果就是每隔10秒续期一次
* scheduleExpirationRenewal(threadId);
* }
* }
* });
* }
* // internalLockLeaseTime=30 * 1000 即30秒
* }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); //30/3=10秒后异步执行续期函数
*
* if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
* task.cancel();
* }
* }
*
* 续期的lua脚本:判断key，field存在则重置过期时间
* protected RFuture<Boolean> renewExpirationAsync(long threadId) {
* return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
* "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
* "redis.call('pexpire', KEYS[1], ARGV[1]); " +
* "return 1; " +
* "end; " +
* "return 0;",
* Collections.<Object>singletonList(getName()),
* internalLockLeaseTime, getLockName(threadId));
* }
*
*
*
* 4，阻塞锁实现
* public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
* long threadId = Thread.currentThread().getId();
* Long ttl = tryAcquire(leaseTime, unit, threadId);
* // lock acquired
* if (ttl == null) {
* return;
* }
* //如果没有获取到锁，则订阅：redisson_lock__channel:{key} 频道
* RFuture<RedissonLockEntry> future = subscribe(threadId);
* commandExecutor.syncSubscription(future);
*
* try {
* while (true) {
* //尝试再获取一次
* ttl = tryAcquire(leaseTime, unit, threadId);
* // lock acquired
* if (ttl == null) {
* break;
* }
*
* // waiting for message 阻塞等待锁订阅频道的消息,一旦锁被释放,就会得到信号通知,继续尝试获取锁
* if (ttl >= 0) {
* getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
* } else {
* getEntry(threadId).getLatch().acquire();
* }
* }
* } finally {
* //获取到锁后取消订阅
* unsubscribe(future, threadId);
* }
* // get(lockAsync(leaseTime, unit));
* }
*
*
* 5，解锁
* protected RFuture<Boolean> unlockInnerAsync(long threadId) {
* return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
* //key已经不存在了，则向redisson_lock__channel:{key}频道发布锁释放消息
* "if (redis.call('exists', KEYS[1]) == 0) then " +
* "redis.call('publish', KEYS[2], ARGV[1]); " +
* "return 1; " +
* "end;" +
* // hash 中的field 不存在时直接返回,
* "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
* "return nil;" +
* "end; " +
* "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
* //重入次数-1后如果还大于0,延长过期时间
* "if (counter > 0) then " +
* "redis.call('pexpire', KEYS[1], ARGV[2]); " +
* "return 0; " +
* "else " +
* //重入次数-1后如果归0,则删除key，并向redisson_lock__channel:{key}频道发布锁释放消息
* "redis.call('del', KEYS[1]); " +
* "redis.call('publish', KEYS[2], ARGV[1]); " +
* "return 1; "+
* "end; " +
* "return nil;",
* Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
*
* }
*/
二、布隆过滤器（BloomFilter）
引言
-
问题1：什么是Redis缓存穿透？缓存穿透如何解决？
-
问题2：如何在海量元素中（例如 10 亿无序、不定长、不重复）快速判断一个元素是否存在？
2.1、什么是 BloomFilter
布隆过滤器 （英语：Bloom Filter）是 1970 年由Burton Howard Bloom提出的，是一种空间效率高的概率型数据结构。
本质上其实就是一个很长的二进制向量和一系列随机映射函数。 专门用来检测集合中是否存在特定的元素 。
2.1.1、产生的契机
平常在检测集合中是否存在某元素时，都会采用比较的方法。考虑以下情况
-
如果集合用线性表存储，查找的时间复杂度为O(n)。
-
如果用平衡BST（如AVL树、红黑树）存储，时间复杂度为O(logn)。
-
如果用哈希表存储，并用链地址法与平衡BST解决哈希冲突（参考JDK8的HashMap实现方法），时间复杂度也要有O[log(n/m)]，m为哈希分桶数。
总而言之，当集合中元素的数量极多时，不仅查找会变得很慢，而且占用的空间也会大到无法想象。BF就是解决这个矛盾的利器。
2.1.2、数据结构&设计思想
BF是由一个长度为m比特的位数组（bit array）与k个哈希函数（hashfunction）组成的数据结构。位数组均初始化为0，所有哈希函数都可以分别把输入数据尽量均匀地散列。
基于BitMap
如果要映射一个值到布隆过滤器中，需要使用 多个不同的哈希函数 生成 多个哈希值 ，并对每个生成的哈希值指向的 bit 位，设置为1。
当要 插入 一个元素时，将其数据分别输入k个哈希函数，产生k个哈希值。以哈希值作为位数组中的下标，将所有k个对应的比特置为1。
当要 查询
（即判断是否存在）一个元素时，同样将其数据输入哈希函数，然后检查对应的k个比特。如果有任意一个比特为0，表明该元素一定不在集合中。如果所有比特均为1，表明该集合有（较大的）可能性在集合中。为什么不是一定在集合中呢？因为一个比特被置为1有可能会受到其他元素的影响，这就是所谓“假阳性”（falsepositive）。相对地，“假阴性”（false negative）在BF中是绝不会出现的。
-
如果这些点有任何一个 0，则被检索元素 一定不在 ；
-
如果都是 1，则被检索元素 很可能在 。
2.1.3、误判率问题分析
哈希函数有以下两个特点
-
如果两个散列值是不相同的（根据同一函数），那么这两个散列值的原始输入也是不相同的。
-
散列函数的输入和输出不是唯一对应关系的，如果两个散列值相同，两个输入值很可能是相同的。但也可能不同，这种情况称为 “散列碰撞”（或者 “散列冲突”）
布隆过滤器的误判是指多个输入经过哈希之后在相同的bit位置1了，这样就无法判断究竟是哪个输入产生的，因此误判的根源在于相同的 bit 位被多次映射且置 1。
2.1.4、不支持删除
hash碰撞这种情况也造成了布隆过滤器的删除问题，传统的布隆过滤器并不支持删除操作，因为布隆过滤器的每一个 bit
并不是独占的，很有可能多个元素共享了某一位。如果直接删除这一位的话，会影响其他的元素。
2.1.5、如何选择哈希函数个数和布隆过滤器长度
很显然，过小的布隆过滤器很快所有的 bit 位均为
1，那么查询任何值都会返回“可能存在”，起不到过滤的目的了。布隆过滤器的长度会直接影响误报率，布隆过滤器越长其误报率越小。
另外，哈希函数的个数也需要权衡，个数越多则布隆过滤器 bit 位置位 1 的速度越快，且布隆过滤器的效率越低；但是如果太少的话，那误报率会变高。
如何选择适合业务的 k 和 m 值呢，这里直接贴一个公式
2.2、布隆过滤器实现
2.2.1、第一种方式： Guava
1、引入Guava pom配置
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>29.0-jre</version>
</dependency>
2、代码实现
public class BloomFilterTest {
@Test
public void test1() {
BloomFilter<Integer> bloomFilter =
BloomFilter.create(Funnels.integerFunnel(), size, fpp);
// 插入10万样本数据
for (int i = 0; i < size; i++) {
bloomFilter.put(i);
}
// 用另外十万测试数据，测试误判率
int count = 0;
for (int i = capacity; i < size + 100000; i++) {
if (bloomFilter.mightContain(i)) {
count++;
System.out.println(i + "误判了");
}
}
System.out.println("总共的误判数:" + count);
}
}
运行结果：
10万数据里有947个误判，约等于0.01%，也就是代码里设置的误判率：fpp = 0.01。
代码分析
核心BloomFilter.create 方法
@VisibleForTesting
static <T> BloomFilter<T> create(Funnel<? super T> funnel, long expectedInsertions, double fpp, Strategy strategy) {
......
}
这里有四个参数
-
funnel ：数据类型(通常是调用Funnels工具类中的)
-
expectedInsertions ：指望插入的值的个数
-
fpp ：误判率(默认值为0.03)
-
strategy ：哈希算法
fpp误判率
-
情景一： fpp = 0.01
误判个数：947 占内存大小：9585058位数 -
情景二： fpp = 0.03 （默认参数）
误判个数：3033 占内存大小：7298440位数
总结
-
误判率能够经过fpp 参数进行调节。
-
fpp越小，须要的内存空间就越大：0.01须要900多万位数，0.03须要700多万位数。
-
fpp越小，集合添加数据时，就须要更多的hash函数运算更多的hash值，去存储到对应的数组下标里（忘了去看上面的布隆过滤存入数据的过程）。
2.2.2、第二种方式：Redisson
上面使用Guava实现的布隆过滤器是把数据放在了本地内存中。分布式的场景中就不合适了，没法共享内存。
还能够用Redis来实现布隆过滤器，这里使用Redis封装好的客户端工具Redisson。
pom配置
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.13.4</version>
</dependency>
Java代码
public class RedissonBloomFilter {
public static void main(String[] args) {
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
config.useSingleServer().setPassword("1234");
//构造Redisson
RedissonClient redisson = Redisson.create(config);
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("phoneList");
//初始化布隆过滤器：预计元素为100000000L,偏差率为3%
bloomFilter.tryInit(100000000L,0.03);
//将号码10086插入到布隆过滤器中
bloomFilter.add("10086");
//判断下面号码是否在布隆过滤器中
//输出false
System.out.println(bloomFilter.contains("123456"));
//输出true
System.out.println(bloomFilter.contains("10086"));
}
}
