wiki

本页面只读。您可以查看源文件，但不能更改它。如果您觉得这是系统错误，请联系管理员。
====== redis实现分布式锁 ======
===== 思路 =====
使用redis的setnx()、get()、getset()方法
  - setnx(lockkey, 当前时间+过期超时时间) ，如果返回1，则获取锁成功；如果返回0则没有获取到锁，转向2。
  - get(lockkey)获取值oldExpireTime ，并将这个value值与当前的系统时间进行比较，如果小于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向3。
  - 计算newExpireTime=当前时间+过期超时时间，然后getset(lockkey, newExpireTime) 会返回当前lockkey的值currentExpireTime。
  - 判断currentExpireTime与oldExpireTime 是否相等，如果相等，说明当前getset设置成功，获取到了锁。如果不相等，说明这个锁又被别的请求获取走了，那么当前请求可以直接返回失败，或者继续重试。
  - 在获取到锁之后，当前线程可以开始自己的业务处理，当处理完毕后，比较自己的处理时间和对于锁设置的超时时间，如果小于锁设置的超时时间，则直接执行delete释放锁；如果大于锁设置的超时时间，则不需要再锁进行处理。
===== 代码 =====
<code java RedisLock.java>
package com.gxx.record.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gxx.record.service.RedisService;
import com.gxx.record.utils.SpringUtils;

/**
 * Redis分布式锁
 * @author Gxx
 */
public class RedisLock {

	/**
	 * 日志记录器
	 */
    private static Logger logger = LoggerFactory.getLogger(RedisLock.class);

    /**
     * redis服务
     */
    private RedisService redisService = (RedisService)SpringUtils.getSpringBeanById("redisService");

    /**
     * 获取不到锁，每隔100毫秒重新获取
     */
    private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;

    /**
     * 锁键路径
     */
    private String lockKey;

    /**
     * 锁超时时间，防止线程在入锁以后，无限的执行等待
     */
    private int expireMsecs = 60 * 1000;

    /**
     * 锁等待时间，防止线程饥饿
     */
    private int timeoutMsecs = 10 * 1000;

    private volatile boolean locked = false;

    /**
     * 构造方法
     * @param lockKey 锁键路径
     */
    public RedisLock(String lockKey) {
    	this.lockKey = "REDIS_LOCK_" + lockKey;
    }

    /**
     * 构造方法
     * @param lockKey 锁键路径
     * @param timeoutMsecs 锁等待时间
     */
    public RedisLock(String lockKey, int timeoutMsecs) {
    	this.lockKey = "REDIS_LOCK_" + lockKey;
        this.timeoutMsecs = timeoutMsecs;
    }

    /**
     * 构造方法
     * @param lockKey 锁键路径
     * @param timeoutMsecs 锁等待时间
     * @param expireMsecs 锁超时时间
     */
    public RedisLock(String lockKey, int timeoutMsecs, int expireMsecs) {
        this.lockKey = "REDIS_LOCK_" + lockKey;
        this.timeoutMsecs = timeoutMsecs;
        this.expireMsecs = expireMsecs;
    }

    /**
     * 获取锁键路径
     * @return
     */
    public String getLockKey() {
        return lockKey;
    }

    /**
     * 获得 lock.
     * 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
     * reids缓存的key是锁的key,所有的共享, value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间)
     * 执行过程:
     * 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
     * 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
     * @return 
     */
    public synchronized boolean lock() throws InterruptedException {
    	logger.info("获取锁开始！");
        int timeout = timeoutMsecs;
        while (timeout >= 0) {
            long expires = System.currentTimeMillis() + expireMsecs + 1;
            String expiresStr = String.valueOf(expires); //锁到期时间
            logger.info("锁到期时间:" + expiresStr);
            logger.info("执行setNX,lockKey=" + lockKey);
            if (redisService.setNX(lockKey, expiresStr)) {
            	logger.info("获取到锁！");
                // lock acquired
                locked = true;
                return true;
            }
            logger.info("没获取到锁");
            logger.info("执行get,lockKey=" + lockKey);
            String currentValueStr = (String)redisService.get(lockKey); //redis里的时间
            if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
            	logger.info("老锁已经失效！");
                logger.info("执行getSet,lockKey=" + lockKey);
                //判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
                // lock is expired
                String oldValueStr = (String)redisService.getSet(lockKey, expiresStr);
                //获取上一个锁到期时间，并设置现在的锁到期时间，
                //只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
                if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                	logger.info("获取到锁！");
                    //防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受

                    //[分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
                    // lock acquired
                    locked = true;
                    return true;
                }
            }
            logger.info("老锁未失效");
            timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;

            /*
                延迟100 毫秒,  这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
                只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进行,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
                使用随机的等待时间可以一定程度上保证公平性
             */
            Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);

        }
        logger.info("超过锁等待时间，没有获取到锁！");
        return false;
    }


    /**
     * 释放lock
     */
    public synchronized void unlock() {
        if (locked) {
        	/**
        	 * 为了让分布式锁的算法更稳键些，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL操作，因为可能客户端因为某个耗时的操作而挂起，
        	 * 操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。 TODO
        	 */
        	logger.info("释放锁,lockKey=" + lockKey);
        	boolean delResult = redisService.delete(lockKey);
        	logger.info("删除结束！delResult=" + delResult);
            locked = false;
        }
    }

}
</code>

<code java ThreadController.java>
/**
	 * 测试分布式锁RedisLock
	 * @param request
	 * @return
	 */
	@RequestMapping(value = "/testRedisLock",produces="application/json")
	public @ResponseBody String testRedisLock() {
		logger.info("测试RedisLock，开始===");
		RedisLock redisLock = new RedisLock("TEST_LOCK", 5*1000, 30*1000);
		try {
			if(redisLock.lock()) {
				logger.info("睡眠15秒开始");
				Thread.sleep(15*1000);
				logger.info("睡眠15秒结束");
			}
		} catch (InterruptedException e) {
			logger.error("异常发生！", e);
		} finally {
			redisLock.unlock();
		}
		logger.info("测试RedisLock，结束===");
		return "testRedisLockOK~";
	}
</code>
wiki

用户工具

站点工具

页面工具
