详解Java如何实现基于Redis的分布式锁_java

前言

单JVM内同步好办， 直接用JDK提供的锁就可以了，但是跨进程同步靠这个肯定是不可能的，这种情况下肯定要借助第三方，我这里实现用Redis，当然还有很多其他的实现方式。其实基于Redis实现的原理还算比较简单的，在看代码之前建议大家先去看看原理，看懂了之后看代码应该就容易理解了。

我这里不实现JDK的java.util.concurrent.locks.Lock接口，而是自定义一个，因为JDK的有个newCondition方法我这里暂时没实现。这个Lock提供了5个lock方法的变体，可以自行选择使用哪一个来获取锁，我的想法是最好用带超时返回的那几个方法，因为不这样的话，假如redis挂了，线程永远都在那死循环了(关于这里，应该还可以进一步优化，如果redis挂了，Jedis的操作肯定会抛异常之类的，可以定义个机制让redis挂了的时候通知使用这个lock的用户，或者说是线程)

package cc.lixiaohui.lock;

import java.util.concurrent.TimeUnit;

public interface Lock {

 /**
 * 阻塞性的获取锁, 不响应中断
 */
 void lock;

 /**
 * 阻塞性的获取锁, 响应中断
 *
 * @throws InterruptedException
 */
 void lockInterruptibly throws InterruptedException;

 /**
 * 尝试获取锁, 获取不到立即返回, 不阻塞
 */
 boolean tryLock;

 /**
 * 超时自动返回的阻塞性的获取锁, 不响应中断
 *
 * @param time
 * @param unit
 * @return {@code true} 若成功获取到锁, {@code false} 若在指定时间内未���取到锁
  *
 */
 boolean tryLock(long time, TimeUnit unit);

 /**
 * 超时自动返回的阻塞性的获取锁, 响应中断
 *
 * @param time
 * @param unit
 * @return {@code true} 若成功获取到锁, {@code false} 若在指定时间内未获取到锁
 * @throws InterruptedException 在尝试获取锁的当前线程被中断
 */
 boolean tryLockInterruptibly(long time, TimeUnit unit) throws InterruptedException;

 /**
 * 释放锁
 */
 void unlock;

}

看其抽象实现：

package cc.lixiaohui.lock;

import java.util.concurrent.TimeUnit;

/**
 * 锁的骨架实现, 真正的获取锁的步骤由子类去实现.
 *
 * @author lixiaohui
 *
 */
public abstract class AbstractLock implements Lock {

 /**
 * <pre>
 * 这里需不需要保证可见性值得讨论, 因为是分布式的锁,
 * 1.同一个jvm的多个线程使用不同的锁对象其实也是可以的, 这种情况下不需要保证可见性
 * 2.同一个jvm的多个线程使用同一个锁对象, 那可见性就必须要保证了.
 * </pre>
 */
 protected volatile boolean locked;

 /**
 * 当前jvm内持有该锁的线程(if have one)
 */
 private Thread exclusiveOwnerThread;

 public void lock {
 try {
 lock(false, 0, null, false);
 } catch (InterruptedException e) {
 // TODO ignore
 }
 }

 public void lockInterruptibly throws InterruptedException {
 lock(false, 0, null, true);
 }

 public boolean tryLock(long time, TimeUnit unit) {
 try {
 return lock(true, time, unit, false);
 } catch (InterruptedException e) {
 // TODO ignore
 }
 return false;
 }

 public boolean tryLockInterruptibly(long time, TimeUnit unit) throws InterruptedException {
 return lock(true, time, unit, true);
 }

 public void unlock {
 // TODO 检查当前线程是否持有锁
 if (Thread.currentThread != getExclusiveOwnerThread) {
 throw new IllegalMonitorStateException("current thread does not hold the lock");
 }

 unlock0;
 setExclusiveOwnerThread(null);
 }

 protected void setExclusiveOwnerThread(Thread thread) {
 exclusiveOwnerThread = thread;
 }

 protected final Thread getExclusiveOwnerThread {
 return exclusiveOwnerThread;
 }

 protected abstract void unlock0;

 /**
 * 阻塞式获取锁的实现
 *
 * @param useTimeout
 * @param time
 * @param unit
 * @param interrupt 是否响应中断
 * @return
 * @throws InterruptedException
 */
 protected abstract boolean lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt) throws InterruptedException;

}

基于Redis的最终实现，关键的获取锁，释放锁的代码在这个类的lock方法和unlock0方法里，大家可以只看这两个方法然后完全自己写一个：

package cc.lixiaohui.lock;

import java.util.concurrent.TimeUnit;

import redis.clients.jedis.Jedis;

/**
 * <pre>
 * 基于Redis的SETNX操作实现的分布式锁
 *
 * 获取锁时最好用lock(long time, TimeUnit unit), 以免网路问题而导致线程一直阻塞
 *
 * <a href="http://redis.io/commands/setnx">SETNC操作参考资料</a>
 * </pre>
 *
 * @author lixiaohui
 *
 */
public class RedisBasedDistributedLock extends AbstractLock {

 private Jedis jedis;

 // 锁的名字
 protected String lockKey;

 // 锁的有效时长(毫秒)
 protected long lockExpires;

 public RedisBasedDistributedLock(Jedis jedis, String lockKey, long lockExpires) {
 this.jedis = jedis;
 this.lockKey = lockKey;
 this.lockExpires = lockExpires;
 }

 // 阻塞式获取锁的实现
 protected boolean lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt) throws InterruptedException{
 if (interrupt) {
 checkInterruption;
 }

 long start = System.currentTimeMillis;
 long timeout = unit.toMillis(time); // if !useTimeout, then it's useless

 while (useTimeout ? isTimeout(start, timeout) : true) {
 if (interrupt) {
 checkInterruption;
 }

 long lockExpireTime = System.currentTimeMillis + lockExpires + 1;//锁超时时间
 String stringOfLockExpireTime = String.valueOf(lockExpireTime);

 if (jedis.setnx(lockKey, stringOfLockExpireTime) == 1) { // 获取到锁
 // TODO 成功获取到锁, 设置相关标识
 locked = true;
 setExclusiveOwnerThread(Thread.currentThread);
 return true;
 }

 String value = jedis.get(lockKey);
 if (value != null && isTimeExpired(value)) { // lock is expired
 // 假设多个线程(非单jvm)同时走到这里
 String oldValue = jedis.getSet(lockKey, stringOfLockExpireTime); // getset is atomic
 // 但是走到这里时每个线程拿到的oldValue肯定不可能一样(因为getset是原子性的)
 // 加入拿到的oldValue依然是expired的，那么就说明拿到锁了
 if (oldValue != null && isTimeExpired(oldValue)) {
  // TODO 成功获取到锁, 设置相关标识
  locked = true;
  setExclusiveOwnerThread(Thread.currentThread);
  return true;
 }
 } else {
 // TODO lock is not expired, enter next loop retrying
 }
 }
 return false;
 }

 public boolean tryLock {
 long lockExpireTime = System.currentTimeMillis + lockExpires + 1;//锁超时时间
 String stringOfLockExpireTime = String.valueOf(lockExpireTime);

 if (jedis.setnx(lockKey, stringOfLockExpireTime) == 1) { // 获取到锁
 // TODO 成功获取到锁, 设置相关标识
 locked = true;
 setExclusiveOwnerThread(Thread.currentThread);
 return true;
 }

 String value = jedis.get(lockKey);
 if (value != null && isTimeExpired(value)) { // lock is expired
 // 假设多个线程(非单jvm)同时走到这里
 String oldValue = jedis.getSet(lockKey, stringOfLockExpireTime); // getset is atomic
 // 但是走到这里时每个线程拿到的oldValue肯定不可能一样(因为getset是原子性的)
 // 假如拿到的oldValue依然是expired的，那么就说明拿到锁了
 if (oldValue != null && isTimeExpired(oldValue)) {
 // TODO 成功获取到锁, 设置相关标识
 locked = true;
 setExclusiveOwnerThread(Thread.currentThread);
 return true;
 }
 } else {
 // TODO lock is not expired, enter next loop retrying
 }

 return false;
 }

 /**
 * Queries if this lock is held by any thread.
 *
 * @return {@code true} if any thread holds this lock and
  *   {@code false} otherwise
 */
 public boolean isLocked {
 if (locked) {
 return true;
 } else {
 String value = jedis.get(lockKey);
 // TODO 这里其实是有问题的, 想:当get方法返回value后, 假设这个value已经是过期的了,
 // 而就在这瞬间, 另一个节点set了value, 这时锁是被别的线程(节点持有), 而接下来的判断
 // 是检测不出这种情况的.不过这个问题应该不会导致其它的问题出现, 因为这个方法的目的本来就
 // 不是同步控制, 它只是一种锁状态的报告.
 return !isTimeExpired(value);
 }
 }

 @Override
 protected void unlock0 {
 // TODO 判断锁是否过期
 String value = jedis.get(lockKey);
 if (!isTimeExpired(value)) {
 doUnlock;
 }
 }

 private void checkInterruption throws InterruptedException {
 if(Thread.currentThread.isInterrupted) {
 throw new InterruptedException;
 }
 }

 private boolean isTimeExpired(String value) {
 return Long.parseLong(value) < System.currentTimeMillis;
 }

 private boolean isTimeout(long start, long timeout) {
 return start + timeout > System.currentTimeMillis;
 }

 private void doUnlock {
 jedis.del(lockKey);
 }

}

如果将来还换一种实现方式（比如zookeeper之类的），到时直接继承AbstractLock并实现lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt)， unlock0方法即可(所谓抽象嘛)

测试

模拟全局ID增长器，设计一个IDGenerator类，该类负责生成全局递增ID，其代码如下：

package cc.lixiaohui.lock;

import java.math.BigInteger;
import java.util.concurrent.TimeUnit;

/**
 * 模拟ID生成
 * @author lixiaohui
 *
 */
public class IDGenerator {

 private static BigInteger id = BigInteger.valueOf(0);

 private final Lock lock;

 private static final BigInteger INCREMENT = BigInteger.valueOf(1);

 public IDGenerator(Lock lock) {
 this.lock = lock;
 }

 public String getAndIncrement {
 if (lock.tryLock(3, TimeUnit.SECONDS)) {
 try {
 // TODO 这里获取到锁, 访问临界区资源
 return getAndIncrement0;
 } finally {
 lock.unlock;
 }
 }
 return null;
 //return getAndIncrement0;
 }

 private String getAndIncrement0 {
 String s = id.toString;
 id = id.add(INCREMENT);
 return s;
 }
}

测试主逻辑：同一个JVM内开两个线程死循环地（循环之间无间隔，有的话测试就没意义了）获取ID（我这里并不是死循环而是跑20s），获取到ID存到同一个Set里面，在存之前先检查该ID在set中是否存在，如果已存在，则让两个线程都停止。如果程序能正常跑完20s，那么说明这个分布式锁还算可以满足要求，如此测试的效果应该和不同JVM(也就是真正的分布式环境中)测试的效果是一样的，下面是测试类的代码：

package cc.lixiaohui.DistributedLock.DistributedLock;

import java.util.HashSet;
import java.util.Set;

import org.junit.Test;

import redis.clients.jedis.Jedis;
import cc.lixiaohui.lock.IDGenerator;
import cc.lixiaohui.lock.Lock;
import cc.lixiaohui.lock.RedisBasedDistributedLock;

public class IDGeneratorTest {

 private static Set<String> generatedIds = new HashSet<String>;

 private static final String LOCK_KEY = "lock.lock";
 private static final long LOCK_EXPIRE = 5 * 1000;

 @Test
 public void test throws InterruptedException {
 Jedis jedis1 = new Jedis("localhost", 6379);
 Lock lock1 = new RedisBasedDistributedLock(jedis1, LOCK_KEY, LOCK_EXPIRE);
 IDGenerator g1 = new IDGenerator(lock1);
 IDConsumeMission consume1 = new IDConsumeMission(g1, "consume1");

 Jedis jedis2 = new Jedis("localhost", 6379);
 Lock lock2 = new RedisBasedDistributedLock(jedis2, LOCK_KEY, LOCK_EXPIRE);
 IDGenerator g2 = new IDGenerator(lock2);
 IDConsumeMission consume2 = new IDConsumeMission(g2, "consume2");

 Thread t1 = new Thread(consume1);
 Thread t2 = new Thread(consume2);
 t1.start;
 t2.start;

 Thread.sleep(20 * 1000); //让两个线程跑20秒

 IDConsumeMission.stop;

 t1.join;
 t2.join;
 }

 static String time {
 return String.valueOf(System.currentTimeMillis / 1000);
 }

 static class IDConsumeMission implements Runnable {

 private IDGenerator idGenerator;

 private String name;

 private static volatile boolean stop;

 public IDConsumeMission(IDGenerator idGenerator, String name) {
 this.idGenerator = idGenerator;
 this.name = name;
 }

 public static void stop {
 stop = true;
 }

 public void run {
 System.out.println(time + ": consume " + name + " start ");
 while (!stop) {
 String id = idGenerator.getAndIncrement;
 if(generatedIds.contains(id)) {
  System.out.println(time + ": duplicate id generated, id = " + id);
  stop = true;
  continue;
 } 

 generatedIds.add(id);
 System.out.println(time + ": consume " + name + " add id = " + id);
 }
 System.out.println(time + ": consume " + name + " done ");
 }

 }

}

说明一点，我这里停止两个线程的方式并不是很好，我是为了方便才这么做的，因为只是测试，最好不要这么做。

测试结果

跑20s打印的东西太多，前面打印的被clear了，只有差不多跑完的时候才有，下面截图。说明了这个锁能正常工作：

当IDGererator没有加锁(即IDGererator的getAndIncrement方法内部获取id时不上锁)时,测试是不通过的，非常大的概率中途就会停止，下面是不加锁时的测试结果:

这个1秒都不到：

这个也1秒都不到：

结束语

好了，以上就是Java实现基于Redis的分布式锁的全部内容，各位如果发现问题希望能指正，希望这篇文章能对大家的学习和工作带来一定的帮助，如果有疑问可以留言交流。

以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索java
， 分布式
， redis
分布式锁
redis实现分布式锁、redis分布式缓存实现、redis如何实现分布式、redis实现分布式事务、用redis实现分布式锁，以便于您获取更多的相关知识。

时间： 2024-10-26 10:56:31

详解Java如何实现基于Redis的分布式锁_java的相关文章

基于Redis的分布式锁真的安全吗？（下）

自从我写完这个话题的上半部分之后,就感觉头脑中出现了许多细小的声音,久久挥之不去.它们就像是在为了一些鸡毛蒜皮的小事而相互争吵个不停.的确,有关分布式的话题就是这样,琐碎异常,而且每个人说的话听起来似乎都有道理.   今天,我们就继续探讨这个话题的后半部分.本文中,我们将从Antirez反驳Martin Kleppmann的观点开始讲起,然后会涉及到Hacker News上出现的一些讨论内容,接下来我们还会讨论到基于Zookeeper和Chubby的分布式锁是怎样的,并和Redlock进行一些对

基于Redis的分布式锁真的安全吗？（上）

网上有关Redis分布式锁的文章可谓多如牛毛了,不信的话你可以拿关键词"Redis 分布式锁"随便到哪个搜索引擎上去搜索一下就知道了.这些文章的思路大体相近,给出的实现算法也看似合乎逻辑,但当我们着手去实现它们的时候,却发现如果你越是仔细推敲,疑虑也就越来越多.   实际上,大概在一年以前,关于Redis分布式锁的安全性问题,在分布式系统专家Martin Kleppmann和Redis的作者antirez之间就发生过一场争论.由于对这个问题一直以来比较关注,所以我前些日子仔细阅读了与这

基于Redis实现分布式锁

背景在很多互联网产品应用中,有些场景需要加锁处理,比如:秒杀,全局递增ID,楼层生成等等.大部分的解决方案是基于DB实现的,Redis为单进程单线程模式,采用队列模式将并发访问变成串行访问,且多客户端对Redis的连接并不存在竞争关系.其次Redis提供一些命令SETNX,GETSET,可以方便实现分布式锁机制. Redis命令介绍使用Redis实现分布式锁,有两个重要函数需要介绍 SETNX命令(SET if Not eXists)语法:SETNX key value功能:当且仅当 key 不

基于Redis实现分布式锁以及任务队列_Redis

一.前言 双十一刚过不久,大家都知道在天猫.京东.苏宁等等电商网站上有很多秒杀活动,例如在某一个时刻抢购一个原价1999现在秒杀价只要999的手机时,会迎来一个用户请求的高峰期,可能会有几十万几百万的并发量,来抢这个手机,在高并发的情形下会对数据库服务器或者是文件服务器应用服务器造成巨大的压力,严重时说不定就宕机了,另一个问题是,秒杀的东西都是有量的,例如一款手机只有10台的量秒杀,那么,在高并发的情况下,成千上万条数据更新数据库(例如10台的量被人抢一台就会在数据集某些记录下 减1),那次这个

详解Java正则表达式中Pattern类和Matcher类_java

前言 本文将介绍Java正则表达式中的Pattern类与Matcher类.首先我们要清楚指定为字符串的正则表达式必须首先被编译为pattern类的实例.因此如何更好的了解这两个类,是编程人员必须知道的. 以下我们就分别来看看这两个类: 一.捕获组的概念 捕获组可以通过从左到右计算其开括号来编号,编号是从1 开始的.例如,在表达式 ((A)(B(C)))中,存在四个这样的组: 1 ((A)(B(C))) 2 (A) 3 (B(C)) 4 (C) 组零始终代表整个表达式. 以 (?) 开头的组是纯的

详解JAVA常用的时间操作【实用】_java

项目中经常有对时间进行处理的需求,下面是一些常用的操作整理,方便以后再次使用以及做相关复习. 1.字符串转换为日期 /** * 字符串转换为日期 * @param dateStr 需要转换的日期 * @param dateFormat 日期格式yyyy-MM-dd/yyyy-MM-dd HH:mm:ss */ public static Date toDate(String dateStr, SimpleDateFormat dateFormat) throws ParseException{

详解Java向服务端发送文件的方法_java

本文实例为大家分享了Java向服务端发送文件的方法,供大家参考,具体内容如下 /* *给服务端发送文件,主要是IO流. */ import java.io.*; import java.net.*; class send2 { public static void main(String[] args) throws Exception { Socket s = new Socket("192.168.33.1",10005);//建立服务 BufferedReader bufr =

详解Java的设计模式编程中的原型模式_java

定义:用原型实例指定创建对象的种类,并通过拷贝这些原型创建新的对象.类型:创建类模式类图: 原型模式主要用于对象的复制,它的核心是就是类图中的原型类Prototype.Prototype类需要具备以下两个条件: 实现Cloneable接口.在java语言有一个Cloneable接口,它的作用只有一个,就是在运行时通知虚拟机可以安全地在实现了此接口的类上使用clone方法.在java虚拟机中,只有实现了这个接口的类才可以被拷贝,否则在运行时会抛出CloneNotSupportedException

详解Java中的时区类TimeZone的用法_java

一.TimeZone 简介TimeZone 表示时区偏移量,也可以计算夏令时. 在操作 Date, Calendar等表示日期/时间的对象时,经常会用到TimeZone:因为不同的时区,时间不同. 下面说说TimeZone对象的 2种常用创建方式.1.获取默认的TimeZone对象使用方法: TimeZone tz = TimeZone.getDefault() 2.使用 getTimeZone(String id) 方法获取TimeZone对象使用方法: // 获取 "GMT+08:00&qu