大家好呀，我是狂野君，最近总听说茅台的很火，正好今日闲来无事，准备自食其力，自己搞瓶茅台尝尝，怎么搞呢，当然是靠技术了，哈哈哈
好了，废话不多说了，上干多线程并发货了
在并发编程中，我们通过锁，来避免由于Java竞争而造成的数据不一致问题。通常，我们以synchronized 、Loc开源中国k来使用它（单机情况）
我们来看一个案例：
高并发下单超卖问题
@Autowired
RedisTemplate<String,String> redisTemplate;
String maotai = "maotai20210321001";//茅台商品编号
@PostConstruct
public void init(){
//此处模拟向缓存中存入商品库存操作
redisTemplate.opsForValue().set(maotai,"100");
}
@GetMapping("/get/maotai2")
public String seckillMaotai2() {
synchronized (this) {
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";
}
}
}
问题分析：
- 现象：本地锁多线程并发在多节点下失效(集群/分布式)
- 原因：本地锁它只java培训能锁住本地JVM进程中的多个线程，对于多个JVM进程的不同多线程的实现方式线程多线程并发间是锁不住的
- 解决：分布式锁（在分布式环境下提供锁服务，并且达到本地锁的效果）
何为分布式锁
- 当在分布式架构下，数据只有一份（或有限制），此时需要利用锁的技术控制某一时刻修改数据的进程数。
- 用一个状态值表示锁，对锁的占用和释放通过状态值来标识。
分布式锁特点
- 互斥性：不仅要在同一jvm进程下的不同线程间互斥，更要在不面试自我介绍一分钟同jvm进程下的不同线程间互斥
- 锁超时：支持锁的自动释放，防止死锁
- 正确，开源代码网站github高效，高可用：解铃还须系铃人（加锁和解锁必须是同一个线程）开源，加锁和解锁操作一定要高效，提供锁的服务要具备容错性
- 可重入：如果一个线程拿到了锁之后继续去获取锁还能获开源取到，我们称锁是可重入的（方法的递归调用）
- 阻塞/非阻塞：如果获取不到直接返回视为非阻塞的，如果获取不到会等待锁的释放直到获取锁或者等待超时，视为阻塞的
- 公平/非公平：按照请求的顺序获取锁视为公平的
基于Redis实现分布式锁
实现思面试路：
锁的实现主要基于redis的SETNX命令：
SETNX key value
将 key 的值设为 value ，当且仅当 key 不存在。
若给定的开源节流 key 已经存在，则 SETNX 不做任何动作。
SET面试问题NX 是『S多线程面试题ET if Not eXists』(如果不存在，则 SET)java是什么意思的简写。
返回值： 设置成功，返回 1 设置失败，返回 0
使用SET多线程下载NX完java模拟器成同步锁的流程及事项如下：
- 使用
SETNX命令获取锁，若返回0（key已存在，锁已存在）则获取失败，反之面试自我介绍3分钟通用获取成功 - 为了防止获取锁后程序出现异常，导致其他线程/进程调用
SETNX命令总是返回0而进入java语言死锁状态，需要为该key设置一个“合理”的过期时间 - 释放锁，使用
D面试常见问题及回答技巧EL命令将锁java面试题数据删除
实现代码开源阅读版本1：
@GetMapping("/get/maotai3")
public String seckillMaotai3() {
//获取锁
Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockey, "1");
if (islock) {
//设置锁的过期时间
redisTemplate.expire(lockey,5, TimeUnit.SECONDS);
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 {
//释放锁
redisTemplate.delete(lockey);
}
}
return "dont get lock";
}
问题分析：
-
- setnx 和 expi面试技巧re是非原子性操作（解决：2.6以前可用使用lua脚本，2.6以后可用set命令）
- 2.错误解锁（如何保证解铃还须系铃人：给锁加一个唯一标识）
错误解锁问题解决：
@GetMapping("/get/maotai4")
public String seckillMaotai4() {
String requestid = 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 eval = redisConnection.eval(
locklua.getBytes(),
ReturnType.BOOLEAN,
1,
lockey.getBytes(),
requestid.getBytes(),
"5".getBytes()
);
return eval;
}
});*/
//获取锁
Boolean islock = redisTemplate.opsForValue().setIfAbsent(lockey,requestid,5,TimeUnit.SECONDS);
if (islock) {
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 {
//释放锁
//判断是自己的锁才能去释放 这种操作不是原子性的
/*String id = redisTemplate.opsForValue().get(lockey);
if (id !=null && id.equals(requestid)) {
redisTemplate.delete(lockey);
}*/
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";
}
锁续期/锁续命
/**
* 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";
}
锁的可重入/阻塞锁（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 "";
}
redisson
概述
Redisson内置了一系列的分布式对象，分布式集合，分布式锁，分布式服务等诸多功能特性，是一款基于Redis实现，拥有一系列分布式系统功能特性的工具包，是实现分布式系统架构中缓存中间件的最佳选择。
下载地址：github.com/redisson/re…
实现
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.8.2</version>
</dependency>
@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,加锁的（是否支持重入）
2，锁续期的
3，阻塞获取
4，释放
/**
* 源码如下：
* 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 亿无序、不定长、不重复）快速判断一个元素是否存在？
什么是 BloomFilter
布隆过滤器（英语java语言：Bloom Filter）是 1970 年由Burton HowaJavard Bloom提出的，是一种空间效率高的概率型数据结构。
本质上其实就是一个很长的二进制向量和一系列随机映射函数。专门用来检测集合中是否存多线程应用场景例子在特定的元素
产生的契机
回想一下，我们平常在检测集合中是否存在某元素时，都会采开源代码网站github用比较的方法。考虑以下情况：
- 如果集合用线性表javascript存储，查找的时多线程并发中线程的状态间复开源节流杂度为O(njava是什么意思)。
- 如果java编译器用平衡BST（如AV多线程L树、红黑树）存储，时间复杂度为O(logn)。
- 如果用哈希表存储，并用链地址法与平衡BST解决面试自我介绍简单大方哈希冲突（参考JDK8的Has开源hMap实现方法），时间复杂度也要有java怎么读O[log(n/m)]，m为哈希分桶数。
总而言之，当集合中元素的数量极多时，不仅查面试找会变得很慢，而且占用的空间也会大到无法想象。BF就是解决这个矛盾的利器。
数据结构&设计java是什么意思思想
BF是由一个长度为m比特的位数组（bit array） 与k个哈希函多线程是什么数（hashJava function） 组成的数据结构。位数组均初始化为0，所有哈希函数都可以分别把输入数据尽量均匀地散列。
基于BitMap:
如果我们要映射一个值到布隆过滤多线程是什么意思器中，我们需要使用多个不同的哈希多线程函数生成多个哈希值， 并对每个生成的哈希值指向的 bit 位，设置为多线程的实现方式1
例：
当要插入一个元素时，将其数据分别输入k个哈希函数，产生k个哈希值。以哈希值作为位数组中的下标，将所有k个多线程面试题对应的比特置为1开源节流。
当要查询（即判断线程阻塞的解决方法是否存在）一个元素时，同样将其面试自我介绍简单大方数据输入哈希开源节流函数，开源矿工然后检查对应的k个比特。如果有任意一个比特为0，表明该元素一定不在集合中。如果所有比特均为1，表明该集合有（较大的）可能性在集合中。为什么不是一定在集合中呢？因为一个比特被置为1有开源可能会受到其他元素的影响，这就是所谓“假面试技巧阳性”（false positive）。相对地，“假阴性”（false negative）在BF中是绝不会出现的。
- 如果这些点有任何一个 0，则被检索元素一定不在；
- 如果都是 1，则被检索元素很java语言可能在。
误判率问题分析
哈希函数有以下两个特点：
- 如果两个散列面试技巧值是不相同的（根据同一函数），那么这两个散列值的原开源节流始输入也是不相同的。
- 散列函数的输入和输出不是唯一对应关系的，如果两个散列值相同，两个输入值很可能是相同的。但也可能不同，这种情况称为 “散列碰撞”（或者 “散列多线程并发中线程的状态冲突”线程阻塞的解决方法）
布隆过滤器的误判是指多个输入经过哈希之后在相同的bit位置1了，这样就无法判断究竟是哪个输入产生的，因此误判的根源在于相同的 bit 位被多次映射且置 1。
不支持删除
hash碰撞这种情况也造成了布隆过滤器的删除问题，传统的布隆过滤器并不支持删除操作，因为布隆过滤器的每一个 bit 并不是独java模拟器占的，很有可能多个元素共享了某一位。如果我们直接删除这一位的话，会影响其他的元素。
如何选择哈希函数个数和布隆过滤器长度
很显然，过小的布隆过滤器很快所有的 bit 位均为 1，那么查询任何面试问题大全及答案大全值都会返回“可能存在”，起不到过滤的目的了。布隆过滤器的长度会直接影响误报率，布隆过滤器越长其误报率越小。
另外，哈希函数的个数也开源软件需要权衡，个开源中国数越多则布隆过滤器 bit 位置位 1 的速度越快，且布隆过滤器的效率越低；但是如果太少的话，那我们的误报率会变高。
如何面试自我介绍简单大方选择适合业务的 k开源矿工 和 m 值呢，这里直接贴一个公式：
布隆过滤器实现
第一种方式：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误判率
情景一：fpp = 0.01
- 误判个数：947 占内存大小：9585058位数
情景二：fpp = 0.03（默认参数）
- 误判个数：3033 占内存大小：72多线程的应用场景98440位数
总结
- 误判率能够经过
fpp参数进行调节 - fpp越小，须要的内存空间就越大：0.01须要900多万位数，0.03须要700多万位数。
- fpp越小，集合添加数据时，就须要更多的hash函数运算更多的hash值，去存储到对应的数组下标里。（忘了去看上多线程编程面的布隆过滤存入数据的过程）
第二种方式：Redisson
上面使用Gua面试自我介绍简单大方va实现的布隆过滤器是把数据放在了本地内存中。分布式的场景中就不合适了，没法共享内存
还能够用Redis来实现布隆过滤器多线程编程，这里使用Re多线程应用场景例子dis封装好的客户端工多线程是什么具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"));
}
}
Twemproxy
1.1.1 简介
cluster是redis官方提供的集群方案，功能确实强大(在线扩容，缩容等等)，除了官方的cl开源阅读app下载安装uster，业界有很多三方的缓存代理中间件，比如： predixy， codis， redis-cerberus，squirrel ，cellar act。Twemproxy是使用最广泛、同时也是redis官方所认可的实现方案。
Twemproxy（又称为nutcracker）由Twitter开源。是一个轻量级的Redis和Memcached代理，主要用来减少对后端缓存服务器的连接数。
特点：
mejava模拟器mcached时代可以称王称霸，但随着redis自身发展，尤其高版本cluster出现，已逐渐被javascript弱化
优点：
简单可靠，具备生产级别应用能力
减少了redis连接数，降低redis连接成本，cluster的所有节点之间都需要互java面试题相建立连接。
除了redis，Twemproxy可以对Me多线程的实现方式mcached 协议做代理，在缓存界是个通用性的解决方案java面试题。
缺点：
和cluster相比，性能有一定javaee的损失（twitter测试约2面试0%）
自身也会成为一个单点，所以，做双活很重要！
它只是一个代理转发，底层的主从切换等依然靠redis自身的主从和哨兵或cluster。这一点上cluster已经javaee完虐它
1.1.2 下载与部署
yum install -y autoconf automake libtool
yum remove -y autoconf
export twemproxy_path=/opt/redis/latest/twemproxy/
mkdir -p $twemproxy_path
cd $twemproxy_path
wget ftp://ftp.gnu.org/gnu/autoconf/autoconf-2.69.tar.gz
tar -zxvf autoconf-2.69.tar.gz
cd autoconf-2.69
.configure --prefix=/usr
make && make install
cd $twemproxy_path
wget http://ftp.gnu.org/gnu/automake/automake-1.14.tar.gz
tar -zxvf automake-1.14.tar.gz
cd automake-1.14
.bootstrap.sh
.configure --prefix=/usr
make && make install
cd $twemproxy_path
wget http://ftp.gnu.org/gnu/libtool/libtool-2.4.2.tar.gz
tar -zxvf libtool-2.4.2.tar.gz
cd libtool-2.4.2
.configure --prefix=/usr
make && make install
cd $twemproxy_path
wget https://github.com/twitter/twemproxy/archive/v0.4.1.tar.gz
tar -zxvf v0.4.1.tar.gz
cd twemproxy-0.4.1
.configure --prefix=/usr
make && make install
#编译完，启动文件在src目录中。
1.1.3 配置与启动
1）先准备好两台redis
#将redis.conf拷贝一份，注意以下配置项
#后台启动
daemonize yes
#bind这一行一定要注释掉！允许外部ip连接，否则将来用redis-cli连接操作命令的时候会报一个错误：
#Error: Connection reset by peer
#bind 127.0.0.1 -::1
#启动两个实例，在8081和8082端口
[root@iZ8vb3a9qxofwannyywl6zZ twemproxy]# pwd
/opt/redis/latest/twemproxy
[root@iZ8vb3a9qxofwannyywl6zZ twemproxy]# ..src/redis-server redis.conf --port 8081
[root@iZ8vb3a9qxofwannyywl6zZ twemproxy]# ..src/redis-server redis.conf --port 8082
#确认服务启动成功
[root@iZ8vb3a9qxofwannyywl6zZ twemproxy]# ps aux | grep redis-server
root 18209 0.1 0.0 162492 2680 ? Ssl 13:35 0:00 ..src/redis-server 127.0.0.1:8081
root 18217 0.0 0.0 162492 2688 ? Ssl 13:35 0:00 ..src/redis-server 127.0.0.1:8082
2）配置twemproxy
#将yml文件拷贝一份，test.yml，并修改内容为自己的redis地址
[root@iZ8vb3a9qxofwannyywl6zZ conf]# pwd
/opt/redis/latest/twemproxy/twemproxy-0.4.1/conf
[root@iZ8vb3a9qxofwannyywl6zZ conf]# cp nutcracker.yml test.yml
#test.yml文件说明
alpha: #标记，如果多组，就alpha，beta……往上加，参考 nutcracker.yml 样本
listen: 127.0.0.1:22121 # 这组集群暴露的端口，将来连这个
hash: fnv1a_64 #hash散列算法
distribution: modula #分片算法，这里用取模方式，一共三种，后面详细介绍
auto_eject_hosts: true #自动摘除故障节点
redis: true #是不是redis，false则表示memcache
server_retry_timeout: 2000 #每隔2秒判断故障节点是否正常，如果正常则放回一致性hash环
server_failure_limit: 3 #多少次无响应，就从一致性hash环中摘除
#redis实例列表，一定要加别名，否则宕机后更换机器，分片就不一样了
#加了别名后，将用别名做分片节点名，否则用ip加端口权重，一旦ip变更会重新迁移
servers:
- 127.0.0.1:8081:1 redis-1
- 127.0.0.1:8082:1 redis-2
#启动：-d后台启动，-c指定启动文件
[root@iZ8vb3a9qxofwannyywl6zZ conf]# ..src/nutcracker -d -c test.yml
[root@iZ8vb3a9qxofwannyywl6zZ conf]# ps aux | grep nutcracker
root 14601 0.0 0.0 1136 248 pts/0 Ssl+ 09:25 0:00 /usr/sbin/nutcracker -c /opt/nutcracker.yml
3）连接与验证
#连接非常的简单，用redis-cli和直连redis一样
#首先在twemproxy上设置多个key，均成功
[root@iZ8vb3a9qxofwannyywl6zZ ~]# redis-cli -p 22121
127.0.0.1:22121> set a a
OK
127.0.0.1:22121> set b b
OK
127.0.0.1:22121> set c c
OK
127.0.0.1:22121> set d d
OK
#先连redis-1 ， 取到ac， bd取不到
[root@iZ8vb3a9qxofwannyywl6zZ ~]# redis-cli -p 8081
127.0.0.1:8081> get a
"a"
127.0.0.1:8081> get b
(nil)
127.0.0.1:8081> get c
"c"
127.0.0.1:8081> get d
(nil)
127.0.0.1:8081>
#再连redis-2 ， 发现ac不存在，bd在这里，验证集群分片成功！
[root@iZ8vb3a9qxofwannyywl6zZ ~]# redis-cli -p 8082
127.0.0.1:8082> get a
(nil)
127.0.0.1:8082> get b
"b"
127.0.0.1:8082> get c
(nil)
127.0.0.1:8082> get d
"d"
1.1.4 分片策略
1）读写原理
写入时，twemproxy将多个对应的key计面试问题大全及答案大全算hash值路由到对应的后端redis机器。
而要在redis集群多线程应用场景例子中查询对应的key/value时，twemproxy同样计算hash值从对应的后端redis收集过来，然后拼接起来返回给用户。
2）策略
后台的redis或memcached集群可以通过以下几种算法进行key/vjava面试题alue的分配（distribution属性）：
- keta开源阅读ma: 一个实现线程阻塞的解决方法一致性hash算法的开源库
- modula: 通过取模的hash算法来选择一个节点
- random：随机选择一个节点
经java编译器典面试题
Redis6.x 之后为何引入了多线程？
答：
Redis6.0 引入多线程主要是为了提高网络 IO 读写性能（Redi多线程面试题s 的瓶颈并不在多线程面试题 CPU，而在内存和网络。）
虽然，Redis6.0 引入了多线程，但是 Redis 的多线程java语言只是在网络数据的读写这类耗时操作上使用了， 执行命令仍然是单线程顺序执行。因此，你也不需要担心线程安java是什么意思全问题。
Redis6.x多线程的实现机制？
流程简述如下：
- 主线程负责接收建立连接请求，获取 Socket 放入全局等待读处理队列。
- 主线javascript程处理完读事件之后，通过 RR（Round Robin）将这些连接分配给这些 IO 线程。
- 主线程阻塞等待 IO 线程读取 Socket 完毕。
- 主线程通过单线程的方式执行请求命令，请求数据读取并解析完成，但并不执行。
- 主线程阻塞等待 IO 线程将数据回写 Socket 完毕。
- 解除线程阻塞的解决方法绑定，清空等待队列。
该设计有如下特点：
1、IO 线java环境变量配置程要么同时在读 socket，要么同时在写，不会同时读或写
2、IO 线程只负责读写 socket 解析命令，不开源节流是什么意思负责命令处理
Redis6.x默认是否开启了多线程？
Redis6.0 的多线程默认是禁用的，只使用主线程
如需开启需要修改 redis 配置文件 redis.conf ：
io-threads-do-reads yes
开启多线程后，还需要设置线程数，否则是不生效的。同样需要修改 redis 配置文件 r多线程编程edis.conf :
io-threads 4 #官网建议4核的机器建议设置为2或3个线程，8核的建议设置为6个线程
缓存穿透
- 缓存穿透： key面试技巧中对应的缓存数据不存在，面试自我介绍一分钟导致去请求数据库，造成数多线程下载据库的压力倍增的情况
- 缓存击穿： redis过期后的一瞬间，有大量用户请求同一个缓存数据，导致这些请java模拟器求都去请求数据库，造成数据库压力倍增的情，针对一个key而java面试题言
- 缓存雪崩： 缓存服务器宕机或者大量缓存集中某个时间段失效，导致请求全部去到数据库，造成数据库压力倍增的情况，这个是针对多个key而言
缓存穿透
概念：缓存穿透是指缓存和数据库中都没有的数据，而用java编译器户不断发起请求，造成数据库的压力倍面试问题大全及答案大全增的情况
例：发起为id值为 -1 的数据或 id 为特别大不存在的数据
解决方案：
（1）接口层增加校验，比如用户鉴权校验，参开源节流数做校验 比如：id 做基础校验，id <=0的直接拦截
（2）对于像ID为负数的非法请求直接过滤掉，采用布隆过滤器(Bloom F多线程是什么ilte线程阻塞r)
（3javaee）针对在数据库中找不到记录的，我们仍然将该空数据存入面试问题缓存中，当然一般会设java编译器置一个较短的过期时间
缓存雪崩
概念：缓存服务器宕机或者大量缓存集开源是什么意思中某个时间段失效，导致请求全部去到数据库，造成数据库压力倍增的情况，这个是针对多个key而言
解决：
（1）实现缓存高可用，通过redis cluster将热点数据均匀分布在不同的Redis库中也能避免全部失效的问题
（2）批量往Redis存数据的时候，把每个Key的失效时间都加个java怎么读随机值
setRedis（Key，value，time + Math.random() * 10000）；
缓存击穿
概念：redis过期后的一瞬间，有大量用多线程的实现方式户请求同一个缓存数据，多线程的实现方式导致这些请求都去请求数据库，造多线程并发中线程的状态成数据库压力倍增的情，针对一个key而言
缓存击穿与缓存雪崩的区别是这里针对的是某一热开源节流门key缓存，而雪崩针对的是大量缓存的集中失效
解开源阅读决方案
● 设置热点数据永远不过期。
● 使用互斥多线程的实现方式锁(mutex key)
好了，java培训今天就先码到这里了，接着去搬砖了，茅台没抢着，倒多线程的实现方式是学习了不少技术，也是收获颇丰
希望大面试技巧和注意事项家不要吝面试常见问题及回答技巧啬你的小手，给狂野君点个赞，你的认可，是我最大的动力
以后，会持续为大家送上干货的，欢迎大家关注，开源阅读app下载安装以免迷路
以下是往期Redis的干货，欢迎收藏
OH面试自我介绍一分钟 MY GOD，原来微博、微信、购物车、抽奖小程序是这么用数据结构的！
新来的，你说下Redis的持久化机制，哪一种能解决我们遇到的这个业务问题？
怎样才能快速成为一名架构师？
