【redis】分布式锁的基本实现

柔情只为你懂 2022-05-15 01:25 128阅读 0赞

【问题背景】

在分布式场景下，出现了拿到的主键id相同的情况，从而导致物流单单号重复，新建失败。物流单号生成的规则是：物流单号标识字符(\*\*) + 发货站点id + yyyyMMdd形式日期 + 自增id。

所以，利用redis，分布式锁机制，保证生成的物流单号不重复。

【相关命令】

1.  incr：将 key 中储存的数字值增1。如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作。如果值包含错误的类型，或字符串类型的值不能表示为数字，那么返回一个错误。
2.  setnx（SET if Not eXists）: 命令在指定的 key 不存在时，为 key 设置指定的值。
3.  getSet：用于设置指定 key 的值，并返回 key 的旧值。
4.  expire：用于设置 key 的过期时间。key 过期后将不再可用。

【实现原理】

1.  通过setnx(lock\_timeout)实现，如果设置了锁返回1， 已经有值没有设置成功返回0
2.  死锁问题：通过时间来判断是否过期，如果已经过期，获取到过期时间get(lockKey)，  
    然后getset(lock\_timeout)判断是否和get相同，相同则证明已经加锁成功  
    因为可能导致多线程同时执行getset(lock\_timeout)方法，这可能导致多线程都只需getset后，对于判断加锁成功的线程，再加expire(lockKey, LOCK\_TIMEOUT, TimeUnit.MILLISECONDS)过期时间，防止多个线程同时叠加时间，导致锁时效时间翻倍

【基本实现】

我们会将id的值存储在redis中，每次生成物流单号，都会从redis缓存获取本次应该得到的id值，即上一次生成之后+1. 在遇到失败的情况下，我们会尝试加锁，从而保证不会有重复的情况产生。

【关键代码】

/** * redis分布式锁 */
    public class DistributedLockUtil { 
        private DistributedLockUtil() {
        }
    
        public static RedisCacheManager redisCacheManager = new RedisCacheManager();
    
        /** * @param lockName 可以为共享变量名，也可以为方法名，主要是用于模拟锁信息 * @return */
        public static boolean lock(String cacheName, String lockName) {
            System.out.println(Thread.currentThread() + "开始尝试加锁！");
            Long result = redisCacheManager.setnx(cacheName, lockName, String.valueOf(System.currentTimeMillis() + 1000));
            if (result != null && result.intValue() == 1) {
                System.out.println(Thread.currentThread() + "加锁成功！");
                redisCacheManager.expire(cacheName, lockName, 1);
                return true;
            } else {
                String lockValueA = redisCacheManager.get(cacheName, lockName);
                //已过期，替换新值
                if (lockValueA != null && Long.parseLong(lockValueA) <= System.currentTimeMillis()) {
                    redisCacheManager.getSet(cacheName, lockName, String.valueOf(System.currentTimeMillis() + 1000));
                    redisCacheManager.expire(cacheName, lockName, 1);
                    return true;
                }
            }
            return false;
        }
    }

import java.util.ArrayList;
    import java.util.HashSet;
    import java.util.List;
    import java.util.Properties;
    import java.util.Set;
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
    import org.apache.log4j.Logger;
    import org.codehaus.jackson.map.DeserializationConfig;
    import org.codehaus.jackson.map.ObjectMapper;
    import org.jbpm.util.ClassLoaderUtil;
    
    import redis.clients.jedis.HostAndPort;
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisSentinelPool;
    import redis.clients.jedis.Pipeline;
    
    /** * redis缓存管理 */
    public class RedisCacheManager implements CacheManager { 
        private static Logger logger = Logger.getLogger(RedisCacheManager.class);
    
        private ObjectMapper mapper = null;
        private static JedisSentinelPool sentinelPool = null;
    
        public RedisCacheManager() {
    
            if (this.mapper == null) {
                this.mapper = new ObjectMapper();
                this.mapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
            }
            if (sentinelPool == null) {
                this.init();
            }
        }
    
        @Override
        public void init() {
    
            Set sentinels = new HashSet();
            //加载连接池的properties文件
            Properties cacheProPerties = ClassLoaderUtil.getProperties("cluster.properties");
            //像连接池中放入的ip：端口 多个以逗号分割
            String cacheClass = cacheProPerties.getProperty("redisSentinelPool");
            String password = cacheProPerties.getProperty("redis.password");
            String masterName = cacheProPerties.getProperty("redis.master");
            long maxWaitMillis = Long.valueOf(cacheProPerties.getProperty("redis.maxWaitMillis"));
            int maxTotal = Integer.parseInt(cacheProPerties.getProperty("redis.maxTotal"));
    
            String c[] = cacheClass.split(",");
            for (int i = 0; i < c.length; i++) {
                String[] j = c[i].split(":");
                sentinels.add(new HostAndPort(j[0], Integer.parseInt(j[1])).toString());
            }
            //JedisSentinelPool配置参数
            GenericObjectPoolConfig gb = new GenericObjectPoolConfig();
            gb.setMaxWaitMillis(maxWaitMillis);
            // 在获取连接的时候检查有效性, 默认false
            gb.setTestOnBorrow(true);
            // 在空闲时检查有效性, 默认false
            gb.setTestWhileIdle(true);
            //最大连接数（客户端）
            gb.setMaxTotal(maxTotal);
            // 对象空闲多久后逐出, 当空闲时间>该值 且 空闲连接>最大空闲数
            // 时直接逐出,不再根据MinEvictableIdleTimeMillis判断 (默认逐出策略)
            gb.setSoftMinEvictableIdleTimeMillis(-1);
    
            if (StringUtils.isNotEmpty(password)) {
                sentinelPool = new JedisSentinelPool(masterName, sentinels, gb, 100000, password);
            } else {
                sentinelPool = new JedisSentinelPool(masterName, sentinels, gb, 100000);
            }
            Jedis jedis = sentinelPool.getResource();
            logger.info("redis启动+++++++++++++++++++");
        }
    
        /** * 成功返回1，失败返回0 * @param key * @param value * @return */
        public Long setnx(String cacheName, String key, String value) {
            Jedis jedis = null;
            Long result = 0L;
            key = cacheName + ":" + key;
            try {
                jedis = sentinelPool.getResource();
                result = jedis.setnx(key, value);
            } catch (Exception e) {
                sentinelPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                if (jedis != null) {
                    sentinelPool.returnResource(jedis);
                }
            }
            return result;
        }
    
        /** * 先获取key对应的value值，若不存在则返回nil，然后将旧的value更新为新的value。 * @param key * @param value * @return */
        public String getSet(String cacheName, String key, String value) {
            Jedis jedis = null;
            String result = null;
            key = cacheName + ":" + key;
            try {
                jedis = sentinelPool.getResource();
                result = jedis.getSet(key, value);
            } catch (Exception e) {
                sentinelPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                if (jedis != null) {
                    sentinelPool.returnResource(jedis);
                }
            }
            return result;
        }
    
        public Long expire(String cacheName, String key, int seconds) {
            Jedis jedis = null;
            Long result = null;
            key = cacheName + ":" + key;
            try {
                jedis = sentinelPool.getResource();
                result = jedis.expire(key, seconds);
            } catch (Exception e) {
                sentinelPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                if (jedis != null) {
                    sentinelPool.returnResource(jedis);
                }
            }
            return result;
        }
    
        public Long incr(String cacheName, String key) {
    
            Jedis jedis = null;
            Long result = null;
            key = cacheName + ":" + key;
            try {
                jedis = sentinelPool.getResource();
                result = jedis.incr(key);
            } catch (Exception e) {
                sentinelPool.returnBrokenResource(jedis);
                e.printStackTrace();
            } finally {
                if (jedis != null) {
                    sentinelPool.returnResource(jedis);
                }
            }
            return result;
        }
    
    }