Erlo

Redis的分布式锁

2019-02-15 10:01:40 发布   367 浏览  
页面报错/反馈
收藏 点赞


一、锁的作用   

  当多线程执行某一业务时(特别是对数据的更新、新增)等操作,可能就会出现多个线程对同一条数据进行修改。其最终的结果一
定与你期望的结果“不太一样”,这就与需要一把锁来控制线程排排队了 - 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("获取分布式锁失败");
            }
        });
    }
}

 

登录查看全部

参与评论

评论留言

还没有评论留言,赶紧来抢楼吧~~

手机查看

返回顶部

给这篇文章打个标签吧~

棒极了 糟糕透顶 好文章 PHP JAVA JS 小程序 Python SEO MySql 确认