Redis的分布式锁

2019-02-17 01:50:58来源：博客园阅读 ()

一、锁的作用　　

　　当多线程执行某一业务时（特别是对数据的更新、新增）等操作,可能就会出现多个线程对同一条数据进行修改。其最终的结果一
定与你期望的结果“不太一样”，这就与需要一把锁来控制线程排排队了 - java内部为我们提供了解决方案，可以使用synchronized
或Lock等方式来实现。
　　但是在生产过程中，因为性能的关系，多数公司都会采用多台服务器来搭建”分布式”。一条请求过来之后，不一定会打到哪台服务
器上，这就保证不了多台服务器的某一”关键业务”同一时间只会有一条线程进行执行。这时就需要一个“媒介”来充当"锁"这个角色,这
个“媒介”需要满足一些特性才能胜任这个工作：
（1）各个服务器都可以对其进行获取和操作（基于数据库的实现方式）；
（2）高性能的获取和释放锁；
（3）具备非阻塞锁特性，如果获取不到锁则立即返回失败；
（4）具备锁等待的特性，即没有获取到锁将继续等待获取锁；
（5）具有锁失效机制，防止死锁；
（6）具备可重入特性；

二、分布式锁的一种实现　　

　　高性能、失效时间、可重入、非阻塞等特性，Redis都能满足，所以基于Redis实现是解决分布式锁的一种方式。基于Redis的分布式
锁的实现思想：
A、加锁
1、计算开始时间，防止一直获取不到锁而死循环；
2、根据系统名称：业务名称：主键ID来创建一个分布式锁的key；
3、循环获取分布式锁（尝试时间=当前时间-开始时间<5000毫秒）；
4、使用setnx方法获取分布式锁（key,value - 当前系统时间+"$TRUE"，失效时间 - 10毫秒）；
5、如果设置成功则获取分布式锁成功，返回true；
6、如果获取失败则睡5毫秒，继续尝试，直到过了时间；
7、如果时间之内没有获取到分布式所，则返回失败；
B、解锁
1、根据系统号，业务名称，主键拼接分布式锁的key；
2、根据key获取到真实的key（使用keys方法）；
3、判断真实的key是否为存在，是否唯一，如果否则为空；
4、如果key唯一则进行删除；
栗子：

/**
 * Jedis的工具类.
 */
public class JedisUtil {
    private static final Logger LOG = LoggerFactory.getLogger(JedisUtil.class);
    private static Map<String, JedisUtil> uniqueInstance = new HashMap();
    private JedisPool jedisPool;
    private static JedisUtil jedisUtil = null;

    public JedisUtil() {
        super();
    }

    /**
     * 选择需要链接哪个库
     * @param serverName
     */
    public JedisUtil(String serverName) {
        if (StringUtils.isBlank(serverName)) {
            throw new RuntimeException("请指定Redis主或从库!");
        } else {
            //实际使用需从配置文件中获取
            String ip = "127.0.0.1";
            Integer port = 6379;
            String pwd = "test";
            //仅仅是测试用
            String redisPoolMaxActive = "30";
            if ("master".equals(serverName)) {
                //链接到主库
                this.initialPool(ip, port, pwd, redisPoolMaxActive);
            }else{
                //可以链接为别的Redis库
            }
        }
    }

    /**
     * 初始化jedisPool
     * @param ip
     * @param port
     * @param pwd
     * @param redisPoolMaxActive
     */
    private void initialPool(String ip, int port, String pwd, String redisPoolMaxActive) {
        JedisPoolConfig config = new JedisPoolConfig();
        //最大连接数，默认8个
        config.setMaxTotal(256);
        //最大空闲连接数，默认8个
        config.setMaxIdle(256);
        if (StringUtils.isNotBlank(redisPoolMaxActive)) {
            Integer redisPoolMaxActiveInt = Integer.valueOf(Integer.parseInt(redisPoolMaxActive));
            if (redisPoolMaxActiveInt.intValue() > 512) {
                redisPoolMaxActiveInt = Integer.valueOf(512);
            }

            config.setMaxTotal(redisPoolMaxActiveInt.intValue());
            config.setMaxIdle(redisPoolMaxActiveInt.intValue());
        }
        //获取连接时的最大等待毫秒数
        config.setMaxWaitMillis(1000L);
        //在获取连接的时候检查有效性,默认false
        config.setTestOnBorrow(true);
        config.setTestOnReturn(true);
        //在空闲时检查有效性,默认false
        config.setTestWhileIdle(true);
        //每次逐出检查时 逐出的最大数目。如果为负数就是: 1/abs(n), 默认3
        config.setNumTestsPerEvictionRun(-1);
        //逐出扫描的时间间隔(毫秒) 如果为负数,则不运行逐出线程, 默认-1
        config.setTimeBetweenEvictionRunsMillis(30000L);
        //逐出连接的最小空闲时间，默认1800000毫秒(30分钟)
        config.setMinEvictableIdleTimeMillis(720000L);
        //对象空闲多久后逐出,当空闲时间>该值且空闲连接 > 最大空闲数时直接逐出,不再根据MinEvictableIdleTimeMillis判断 (默认逐出策略)
        config.setSoftMinEvictableIdleTimeMillis(360000L);
        this.jedisPool = new JedisPool(config, ip, port, 2000, pwd);
        LogUtils.info(LOG, "jedisPool init finish", new Object[]{"jedisPool", this.jedisPool});
    }

    /**
     * 单例获取jedisPool
     * @param serverName Redis库的名称
     * @return
     */
    public static JedisUtil getInstance(String serverName) {
        jedisUtil = (JedisUtil)uniqueInstance.get(serverName);
        if(jedisUtil == null) {
            Class var1 = JedisUtil.class;
            synchronized(JedisUtil.class) {
                jedisUtil = (JedisUtil)uniqueInstance.get(serverName);
                if(jedisUtil == null) {
                    jedisUtil = new JedisUtil(serverName);
                    uniqueInstance.put(serverName, jedisUtil);
                }
            }
        }
        return jedisUtil;
    }

    public Long setnx(String key, String value, int expireTime, int dbIndex) {
        Jedis jedis = null;

        Long var1;
        try {
            jedis = this.jedisPool.getResource();
            jedis.select(dbIndex);
            Long ret = jedis.setnx(key, value);
            if(ret.longValue() == 1L && expireTime > 0) {
                jedis.expire(key, expireTime);
            }

            var1 = ret;
        } catch (Exception var11) {
            LogUtils.error(LOG, "redis setnx error. ", var11, new Object[0]);
            throw var11;
        } finally {
            if(jedis != null) {
                jedis.close();
            }

        }

        return var1;
    }

    /**
     * 查找key
     * @param sKey
     * @param index
     * @return
     */
    public Set<String> key(String sKey, int index) {
        Jedis jedis = null;

        Set var2;
        try {
            jedis = this.jedisPool.getResource();
            jedis.select(index);
            var2 = jedis.keys(sKey);
        } catch (Exception var8) {
            LogUtils.error(LOG, "redis key error. ", var8, new Object[0]);
            throw var8;
        } finally {
            if(jedis != null) {
                jedis.close();
            }

        }

        return var2;
    }

    /**
     * 删除key
     * @param sKey
     * @param index
     * @return
     */
    public Long del(String sKey, int index) {
        Jedis jedis = null;

        Long var3;
        try {
            jedis = this.jedisPool.getResource();
            jedis.select(index);
            var3 = jedis.del(sKey);
        } catch (Exception var8) {
            LogUtils.error(LOG, "redis del error. ", var8, new Object[0]);
            throw var8;
        } finally {
            if(jedis != null) {
                jedis.close();
            }

        }

        return var3;
    }
}

分布式锁的工具类：

public class LockRedisService {
    private static final Logger LOG = LoggerFactory.getLogger(LockRedisService.class);
    private static final Integer EXPIRE_SECOND = Integer.valueOf(10);
    public LockRedisService() {}

    /**
     * 尝试获取分布式锁
     * @param systemName 系统名称
     * @param business   业务名称
     * @param keyfix     主键
     * @return
     */
    public static Boolean tryLock(String systemName, String business, String keyfix) {
        long start = System.currentTimeMillis();
        String key = String.format("lock:%s:%s:%s", new Object[]{systemName, business, keyfix});
        LogUtils.info(LOG, "尝试获取分布式锁", new Object[]{"key", key});

        do {
            try {
                long ret = JedisUtil.getInstance("master").setnx(key, System.currentTimeMillis() + "$TRUE", EXPIRE_SECOND, 1).longValue();
                if(ret == 1L) {
                    LogUtils.info(LOG, "成功获得分布式锁", new Object[]{"key", key});
                    return Boolean.TRUE;
                }

                Thread.sleep(5L);
            } catch (Exception var8) {
                LogUtils.error(LOG, "获取锁失败", var8, new Object[0]);
            }
        } while(System.currentTimeMillis() - start < 5000L);

        LogUtils.warn(LOG, "获取分布式锁失败", new Object[]{"key", key});
        return Boolean.FALSE;
    }

    /**
     * 解锁
     * @param systemName 系统名称
     * @param business   业务名称
     * @param keyfix     主键
     * @return
     */
    public static Boolean unlock(String systemName, String business, String keyfix) {
        String key = String.format("lock:%s:%s:%s", new Object[]{systemName, business, keyfix});

        try {
            //1为使用哪个Redis库，正常应该写为枚举类
            Set<String> realKeySet = JedisUtil.getInstance("master").key(key,1);
            if(CollectionUtils.isEmpty(realKeySet)) {
                return null;
            } else if(realKeySet.size() > 1) {
                LogUtils.warn(LOG, "lock key 通配符存在多个", new Object[]{"key", key});
                return null;
            } else {
                String realKey = (String)realKeySet.iterator().next();
                JedisUtil.getInstance("master").del(realKey, 1);
            }
        } catch (Exception var5) {
            LogUtils.error(LOG, "解锁失败", var5, new Object[0]);
        }

        LogUtils.info(LOG, "解锁成功", new Object[]{"key", key});
        return Boolean.valueOf(true);
    }
}

测试类：

public class Test {
    public static void main(String[] args) {
        //获取cpu的核心数
        int count = Runtime.getRuntime().availableProcessors();
        //创建线程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(count, count * 10,
                60L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(40960), new ThreadPoolExecutor.AbortPolicy());
        //多线程执行业务
        threadPool.execute(()->{
            String bussiness = "测试分布式锁";
            String key = "TestLock:" + 123;
            //尝试获取分布式锁
            if (LockRedisService.tryLock("测试系统", bussiness, key)) {
                System.out.println("---获取分布式锁成功---");
                try {
                    //可能会抛出异常
                    System.out.println("执行逻辑");
                }catch (Exception e){
                    System.out.println("执行逻辑错误: "+e);
                }finally {
                    //执行完毕后解锁
                    LockRedisService.unlock("测试系统", bussiness, key);
                }
            }else {
                System.out.println("获取分布式锁失败");
            }
        });
    }
}