一个Redis实现的分布式锁

本文涉及的产品
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
简介:   import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.data.redis.

 

 

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.Random;
import java.util.concurrent.TimeUnit;

public class RedisLock implements AutoCloseable {
    private static final Logger LOGGER = LoggerFactory.getLogger(RedisLock.class);
    public static final String REDIS_LOCK = "RedisLock:";


    private static final long DEFAULT_WAIT_LOCK_TIME_OUT = 60;//60s 有慢sql,超时时间设置长一点
    private static final long DEFAULT_EXPIRE = 80;//80s 有慢sql,超时时间设置长一点
    private String key;
    private RedisTemplate redisTemplate;

    public RedisLock(RedisTemplate redisTemplate,String key) {
        this.redisTemplate = redisTemplate;
        this.key = key;
    }

    /**
     * 等待锁的时间,单位为s
     *
     * @param key
     * @param timeout s
     * @param seconds
     */
    public boolean lock(String key, long timeout, TimeUnit seconds) {
        String lockKey = generateLockKey(key);
        long nanoWaitForLock = seconds.toNanos(timeout);
        long start = System.nanoTime();

        try {
            while ((System.nanoTime() - start) < nanoWaitForLock) {
                if (redisTemplate.getConnectionFactory().getConnection().setNX(lockKey.getBytes(), new byte[0])) {
                    redisTemplate.expire(lockKey, DEFAULT_EXPIRE, TimeUnit.SECONDS);//暂设置为80s过期,防止异常中断锁未释放
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("add RedisLock[{}].{}", key, Thread.currentThread());
                    }
                    return true;
                }
                TimeUnit.MILLISECONDS.sleep(1000 + new Random().nextInt(100));//加随机时间防止活锁
            }
        } catch (Exception e) {
            LOGGER.error("{}", e.getMessage(), e);
            unlock();
        }
        return false;
    }

    public void unlock() {
        try {
            String lockKey = generateLockKey(key);
            RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
            connection.del(lockKey.getBytes());
            connection.del(key.getBytes());
            connection.close();
        } catch (Exception e) {
            LOGGER.error("{}", e.getMessage(), e);
        }
    }

    private String generateLockKey(String key) {
        return String.format(REDIS_LOCK + "%s", key);
    }

    public boolean lock() {
        return lock(key, DEFAULT_WAIT_LOCK_TIME_OUT, TimeUnit.SECONDS);
    }

    @Override
    public void close(){
        try {
            String lockKey = generateLockKey(key);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("release RedisLock[" + lockKey + "].");
            }
            RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
            connection.del(lockKey.getBytes());
            connection.close();
        } catch (Exception e) {
            LOGGER.error("{}", e.getMessage(), e);
        }
    }
}

 

 

在高并发的使用场景下,如何让redis里的数据尽量保持一致,可以采用分布式锁。以分布式锁的方式来保证对临界资源的互斥读写。

   redis使用缓存作为分布式锁,性能非常强劲,在一些不错的硬件上,redis可以每秒执行10w次,内网延迟不超过1ms,足够满足绝大部分应用的锁定需求。

   redis常用的分布式锁的实现方式:

一、setbit / getbit

   用索引号为0的第一个比特位来表示锁定状态,其中:0表示未获得锁,1表示已获得锁。

   优势:简单;

   劣势:竞态条件(race condition),死锁。

   获得锁的过程至少需要两步:先getbit判断,后setbit上锁。由于不是原子操作,因此可能存在竞态条件;如果一个客户端使用setbit获取到锁,然后没来得及释放crash掉了,那么其他在等待的客户端将永远无法获得该锁,进而形成了死锁。所以这种形式不太适合实现分布式锁。

二、setnx / del / getset

  redis官网有一篇文章专门谈论了实现分布式锁的话题。基本的原则是:采用setnx尝试获取锁并判断是否获得了锁,setnx设置的值是它想占用锁的时间(预估):

  • 如返回1,则该客户端获得锁,把lock.foo的键值设置为时间值表示该键已被锁定,该客户端最后可以通过DEL lock.foo来释放该锁。
  • 如返回0,表明该锁已被其他客户端取得,这时我们可以先返回或进行重试等对方完成或等待锁超时。

  通过del删除key来释放锁。某个想获得锁的客户端,先采用setnx尝试获取锁,如果获取失败了,那么会通过get命令来获得锁的过期时间以判断该锁的占用是否过期。如果跟当前时间对比,发现过期,那么先执行del,然后执行setnx获取锁。如果整个流程就这样,可能会产生死锁,请参考下面的执行序列:

   所以,在高并发的场景下,如果检测到锁过期,不能简单地进行del并尝试通过setnx获得锁。我们可以通过getset命令来避免这个问题。来看看,如果存在一个用户user4,它通过调用getset命令如何避免这种情况的发生:

 getset设置的过期时间跟上面的setnx设置的相同:

   如果该命令返回的结果跟上一步通过get获得的过期时间一致,则说明这两步之间,没有新的客户端抢占了锁,则该客户端即获得锁。如果该命令返回的结果跟上一步通过get获得的过期时间不一致,则该锁可能已被其他客户端抢先获得,则本次获取锁失败。

   这种实现方式得益于getset命令的原子性,从而有效得避免了竞态条件。并且,通过将比对锁的过期时间作为获取锁逻辑的一部分,从而避免了死锁。

三、setnx / del / expire

   这是使用最多的实现方式:setnx的目的同上,用来实现尝试获取锁以及判断是否获取到锁的原子性,del删除key来释放锁,与上面不同的是,使用redis自带的expire命令来防止死锁(可能出现某个客户端获得了锁,但是crash了,永不释放导致死锁)。这算是一种比较简单但粗暴的实现方式:因为,不管实际的情况如何,当你设置expire之后,它一定会在那个时间点删除key。如何当时某个客户端已获得了锁,正在执行临界区内的代码,但执行时间超过了expire的时间,将会导致另一个正在竞争该锁的客户端也获得了该锁,这个问题下面还会谈到。

  我们来看一下宿舍锁的简单实现很简单:

通过一个while(true),在当前线程上进行阻塞等待,并通过一个计数器进行自减操作,防止永久等待。 

http://www.cnblogs.com/moonandstar08/p/5682822.html

 

多节点的部署中,对锁的控制,参考:

http://www.jeffkit.info/2011/07/1000/

直接贴上代码实现,同上一篇文章一样,都是基于AOP

定义注解,标志切入点:

package com.ns.annotation;

import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface RedisLock {
    /**
     * redis的key
     * @return
     */
    String value();
    /**
     * 持锁时间,单位毫秒,默认一分钟
     */
    long keepMills() default 60000;
    /**
     * 当获取失败时候动作
     */
    LockFailAction action() default LockFailAction.GIVEUP;
    
    public enum LockFailAction{
        /**
         * 放弃
         */
        GIVEUP,
        /**
         * 继续
         */
        CONTINUE;
    }
    /**
     * 睡眠时间,设置GIVEUP忽略此项
     * @return
     */
    long sleepMills() default 1000;
}

 

切面实现:

package com.redis.aop;
import java.lang.reflect.Method;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import com.ns.annotation.RedisLock;
import com.ns.annotation.RedisLock.LockFailAction;
import com.ns.redis.dao.base.BaseRedisDao;
@Aspect
public class RedisLockAspect extends BaseRedisDao<String, Long>{
  private static final Logger log = LoggerFactory.getLogger(RedisLockAspect.class);
  //execution(* com.ns..*(*,..)) and @within(com.ns.annotation.RedisLock)
  
  @Pointcut("execution(* com.ns..*(..)) && @annotation(com.ns.annotation.RedisLock)")
  private void lockPoint(){}
  @Around("lockPoint()")
  public Object arround(ProceedingJoinPoint pjp) throws Throwable{
    MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
    Method method = methodSignature.getMethod();
    RedisLock lockInfo = method.getAnnotation(RedisLock.class);
    boolean lock = false;
    Object obj = null;
    while(!lock){
      long timestamp = System.currentTimeMillis()+lockInfo.keepMills();
      lock = setNX(lockInfo.value(), timestamp);
      //得到锁,已过期并且成功设置后旧的时间戳依然是过期的,可以认为获取到了锁(成功设置防止锁竞争)
      long now = System.currentTimeMillis();
      if(lock || ((now > getLock(lockInfo.value())) && (now > getSet(lockInfo.value(), timestamp)))){
        //得到锁,执行方法,释放锁
        log.info("得到锁...");
        obj = pjp.proceed();
        //不加这一行,对于只能执行一次的定时任务,时间差上不能保证另一个一定正好放弃
        if(lockInfo.action().equals(LockFailAction.CONTINUE)){
          delete(lockInfo.value());
        }
      }else{
        if(lockInfo.action().equals(LockFailAction.CONTINUE)){
          log.info("稍后重新请求锁...");
          Thread.currentThread().sleep(lockInfo.sleepMills());
        }else{
          log.info("放弃锁...");
          break;
        }
      }
    }
    return obj;
  }
  public boolean setNX(String key,Long value){
    return valueOperations.setIfAbsent(key, value);
  }
  public long getLock(String key){
    return valueOperations.get(key);
  }
  public Long getSet(String key,Long value){
    return valueOperations.getAndSet(key, value);
  }
  public void releaseLock(String key){
    delete(key);
  }
}

Python的一个实现

LOCK_TIMEOUT = 3
lock = 0
lock_timeout = 0
lock_key = 'lock.foo'

# 获取锁
while lock != 1:
    now = int(time.time())
    lock_timeout = now + LOCK_TIMEOUT + 1
    lock = redis_client.setnx(lock_key, lock_timeout)
    if lock == 1 or (now > int(redis_client.get(lock_key))) and now > int(redis_client.getset(lock_key, lock_timeout)):
        break
    else:
        time.sleep(0.001)

# 已获得锁
do_job()

# 释放锁
now = int(time.time())
if now < lock_timeout:
    redis_client.delete(lock_key)

http://blog.csdn.net/lihao21/article/details/49104695

以上有些代码只符合我现在的项目场景,根据实际需要进行调整

http://www.tuicool.com/articles/EzaM7by

 

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
7天前
|
缓存 NoSQL 安全
玩转Redis!非常强大的Redisson分布式集合,少写60%代码
Redisson是Java的Redis客户端,提供实时数据平台服务,简化了分布式环境下的数据管理。它包含RList、RSet、RMap等分布式集合,支持ConcurrentMap和Set接口,确保线程安全和数据一致性。例如,RMap实现了本地缓存和监听器功能,允许数据监听和本地加速读取。此外,还提供了RSet的排序和去重功能,以及RQueue和RBlockingQueue等队列实现,支持阻塞操作。通过Redisson,开发者能轻松处理分布式系统的数据同步和操作。
|
8天前
|
存储 缓存 NoSQL
了解Redis,第一弹,什么是RedisRedis主要适用于分布式系统,用来用缓存,存储数据,在内存中存储那么为什么说是分布式呢?什么叫分布式什么是单机架构微服务架构微服务的本质
了解Redis,第一弹,什么是RedisRedis主要适用于分布式系统,用来用缓存,存储数据,在内存中存储那么为什么说是分布式呢?什么叫分布式什么是单机架构微服务架构微服务的本质
|
10天前
|
缓存 NoSQL 关系型数据库
【Redis】 浅谈分布式架构
【Redis】 浅谈分布式架构
|
17天前
|
存储 NoSQL 算法
Redis (分布式锁)
Redis (分布式锁)
197 0
|
30天前
|
监控 NoSQL 算法
探秘Redis分布式锁:实战与注意事项
本文介绍了Redis分区容错中的分布式锁概念,包括利用Watch实现乐观锁和使用setnx防止库存超卖。乐观锁通过Watch命令监控键值变化,在事务中执行修改,若键值被改变则事务失败。Java代码示例展示了具体实现。setnx命令用于库存操作,确保无超卖,通过设置锁并检查库存来更新。文章还讨论了分布式锁存在的问题,如客户端阻塞、时钟漂移和单点故障,并提出了RedLock算法来提高可靠性。Redisson作为生产环境的分布式锁实现,提供了可重入锁、读写锁等高级功能。最后,文章对比了Redis、Zookeeper和etcd的分布式锁特性。
222 16
探秘Redis分布式锁:实战与注意事项
|
30天前
|
NoSQL Java 大数据
介绍redis分布式锁
分布式锁是解决多进程在分布式环境中争夺资源的问题,与本地锁相似但适用于不同进程。以Redis为例,通过`setIfAbsent`实现占锁,加锁同时设置过期时间避免死锁。然而,获取锁与设置过期时间非原子性可能导致并发问题,解决方案是使用`setIfAbsent`的超时参数。此外,释放锁前需验证归属,防止误删他人锁,可借助Lua脚本确保原子性。实际应用中还有锁续期、重试机制等复杂问题,现成解决方案如RedisLockRegistry和Redisson。
|
30天前
|
缓存 NoSQL Java
【亮剑】分布式锁是保证多服务实例同步的关键机制,常用于互斥访问共享资源、控制访问顺序和系统保护,如何使用注解来实现 Redis 分布式锁的功能?
【4月更文挑战第30天】分布式锁是保证多服务实例同步的关键机制,常用于互斥访问共享资源、控制访问顺序和系统保护。基于 Redis 的分布式锁利用 SETNX 或 SET 命令实现,并考虑自动过期、可重入及原子性以确保可靠性。在 Java Spring Boot 中,可通过 `@EnableCaching`、`@Cacheable` 和 `@CacheEvict` 注解轻松实现 Redis 分布式锁功能。
|
30天前
|
NoSQL Redis 微服务
分布式锁_redis实现
分布式锁_redis实现
|
30天前
|
NoSQL Java 关系型数据库
【Redis系列笔记】分布式锁
分布式锁:满足分布式系统或集群模式下多进程可见并且互斥的锁。 分布式锁的核心思想就是让大家都使用同一把锁,只要大家使用的是同一把锁,那么我们就能锁住线程,不让线程进行,让程序串行执行,这就是分布式锁的核心思路
390 2
|
30天前
|
NoSQL Java Redis
redis分布式锁
redis分布式锁