Redis实现分布式锁和等待序列的方法示例
在集群下，经常会因为同时处理发生资源争抢和并发问题，但是我们都知道同步锁 synchronized、 cas、 ReentrankLock这些锁的作用范围都是 JVM，说白了在集群下没啥用。这时我们就需要能在多台 JVM之间决定执行顺序的锁了，现在分布式锁主要有 redis、 Zookeeper实现的，还有数据库的方式，不过性能太差，也就是需要一个第三方的监管。
背景
最近在做一个消费 Kafka消息的时候发现，由于线上的消费者过多，经常会遇到，多个机器同时处理一个主键类型的数据的情况发生，如果最后是执行更新操作的话，也就是一个更新顺序的问题，但是如果恰好都需要插入数据的时候，会出现主键重复的问题。这是生产上不被允许的(因为公司有异常监管的机制，扣分啥的)，这是就需要个分布式锁了，斟酌后用了 Redis的实现方式(因为网上例子多)
分析
redis实现的分布式锁，实现原理是 set方法，因为多个线程同时请求的时候，只有一个线程可以成功并返回结果，还可以设置有效期，来避免死锁的发生，一切都是这么的完美，不过有个问题，在 set的时候，会直接返回结果，成功或者失败，不具有阻塞效果，需要我们自己对失败的线程进程处理，有两种方式
- 丢弃
- 等待重试由于我们的系统需要这些数据，那么只能重新尝试获取。这里使用 redis的 List类型实现等待序列的作用
代码
直接上代码其实直接redis的工具类就可以解决了
packagecom.test
importredis.clients.jedis.Jedis;
importjava.util.Collections;
importjava.util.List;
/**
*@descredis队列实现方式
*@anthor
*@date
**/
publicclassRedisUcUitl{
privatestaticfinalStringLOCK_SUCCESS="OK";
privatestaticfinalStringSET_IF_NOT_EXIST="NX";
privatestaticfinalStringSET_WITH_EXPIRE_TIME="PX";
privatestaticfinalLongRELEASE_SUCCESS=1L;
privateRedisUcUitl(){
}
/**
*logger
**/
/**
*存储redis队列顺序存储在队列首部存入
*
*@paramkey字节类型
*@paramvalue字节类型
*/
publicstaticLonglpush(Jedisjedis,finalbyte[]key,finalbyte[]value){
returnjedis.lpush(key,value);
}
/**
*移除列表中最后一个元素并将改元素添加入另一个列表中，当列表为空时将阻塞连接直到等待超时
*
*@paramsrckey
*@paramdstkey
*@paramtimeout0表示永不超时
*@return
*/
publicstaticbyte[]brpoplpush(Jedisjedis,finalbyte[]srckey,finalbyte[]dstkey,finalinttimeout){
returnjedis.brpoplpush(srckey,dstkey,timeout);
}
/**
*返回制定的key,起始位置的redis数据
*@paramredisKey
*@paramstart
*@paramend-1表示到最后
*@return
*/
publicstaticListlrange(Jedisjedis,finalbyte[]redisKey,finallongstart,finallongend){
returnjedis.lrange(redisKey,start,end);
}
/**
*删除key
*@paramredisKey
*/
publicstaticvoiddelete(Jedisjedis,finalbyte[]redisKey){
returnjedis.del(redisKey);
}
/**
*尝试加锁
*@paramlockKeykey名称
*@paramrequestId身份标识
*@paramexpireTime过期时间
*@return
*/
publicstaticbooleantryGetDistributedLock(Jedisjedis,finalStringlockKey,finalStringrequestId,finalintexpireTime){
Stringresult=jedis.set(lockKey,requestId,SET_IF_NOT_EXIST,SET_WITH_EXPIRE_TIME,expireTime);
returnLOCK_SUCCESS.equals(result);
}
/**
*释放锁
*@paramlockKeykey名称
*@paramrequestId身份标识
*@return
*/
publicstaticbooleanreleaseDistributedLock(Jedisjedis,finalStringlockKey,finalStringrequestId){
finalStringscript="ifredis.call('get',KEYS[1])==ARGV[1]thenreturnredis.call('del',KEYS[1])elsereturn0end";
jedis.eval(script,Collections.singletonList(lockKey),Collections.singletonList(requestId));
returnRELEASE_SUCCESS.equals(result);
}
}
业务逻辑主要代码如下
1.先消耗队列中的
while(true){
//消费队列
try{
//被放入redis队列的数据序列化后的
byte[]bytes=RedisUcUitl.brpoplpush(keyStr.getBytes(UTF_8),dstKeyStr.getBytes(UTF_8),1);
if(bytes==null||bytes.isEmpty()){
//队列中没数据时退出
break;
}
//反序列化对象
MapsingleMap=(Map)ObjectSerialUtil.bytesToObject(bytes);
//塞入唯一的值防止被其他线程误解锁
StringrequestId=UUID.randomUUID().toString();
booleanlockGetFlag=RedisUcUitl.tryGetDistributedLock(keyStr,requestId,100);
if(lockGetFlag){
//成功获取锁进行业务处理
//TODO
//处理完毕释放锁
booleanfreeLock=RedisUcUitl.releaseDistributedLock(keyStr,requestId);
}else{
//未能获得锁放入等待队列
RedisUcUitl.lpush(keyStr.getBytes(UTF_8),ObjectSerialUtil.objectToBytes(param));
}
}catch(Exceptione){
break;
}
}
2.处理最新接到的数据
同样是走尝试获取锁，获取不到放入队列的流程
一般序列化用 fastJson之列的就可以了，这里用的是 JDK自带的，工具类如下
publicclassObjectSerialUtil{
privateObjectSerialUtil(){
//工具类
}
/**
*将Object对象序列化为byte[]
*
*@paramobj对象
*@returnbyte数组
*@throwsException
*/
publicstaticbyte[]objectToBytes(Objectobj)throwsIOException{
ByteArrayOutputStreambos=newByteArrayOutputStream();
ObjectOutputStreamoos=newObjectOutputStream(bos);
oos.writeObject(obj);
byte[]bytes=bos.toByteArray();
bos.close();
oos.close();
returnbytes;
}
/**
*将bytes数组还原为对象
*
*@parambytes
*@return
*@throwsException
*/
publicstaticObjectbytesToObject(byte[]bytes){
try{
ByteArrayInputStreambin=newByteArrayInputStream(bytes);
ObjectInputStreamois=newObjectInputStream(bin);
returnois.readObject();
}catch(Exceptione){
thrownewBaseException("反序列化出错！",e);
}
}
}
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。