devbox
爱喝兽奶帝天荒
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

分布式锁解决方案_分布式锁解决什么问题

作者：爱喝兽奶帝天荒 | 2024-07-01 12:08:35

赞

踩

分布式锁解决什么问题

锁:

      单进程的系统中，存在多线程同时操作一个公共变量，此时需要加锁对变量进行同步操作，保证多线程的操作线性执行消除并发修改。解决的是单进程中的多线程并发问题。

分布式锁:

      只要的应用场景是在集群模式的多个相同服务，可能会部署在不同机器上，解决进程间安全问题，防止多进程同时操作一个变量或者数据库。解决的是多进程的并发问题。

JVM有锁的感念（局限于同一台java虚拟机）

一：分布式锁具备的条件：

           1，在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；

           2，具备可重入特性；

          3，具备锁失效机制，防止死锁；

          4，具备非阻塞锁特性，即如果没有获取到锁直接返回获取锁失败。

二：分布式锁的实现方式

       1，基于缓存（redis）实现分布式锁(共享锁)；

              1.1，redis有很高的性能； 
              1.2，redis命令对此支持较好，实现起来比较方便.                   

  1. /**
  2.  * redis 分布式锁
  3.  * @author reyco
  4.  *
  5.  */
  6. public class DistributedLock {
  7.  
  8. 	private JedisPool jedisPool;
  9.  
  10. 	public DistributedLock(JedisPool jedisPool) {
  11. 		super();
  12. 		this.jedisPool = jedisPool;
  13. 	}
  14. 	
  15. 	/**
  16. 	 * 获取锁
  17. 	 * 
  18. 	 * @param lockName
  19. 	 *            锁的名称
  20. 	 * @param acquireTimeout
  21. 	 *            获取锁的超时时间,单位毫秒数
  22. 	 * @param timeout
  23. 	 *            key的超时时间,单位毫秒数
  24. 	 */
  25. 	public String lock(String lockName, long acquireTimeout, long timeout) {
  26. 		Jedis jedis = null;
  27. 		// 返回标识
  28. 		String identifier = null;
  29. 		try {
  30. 			jedis = jedisPool.getResource();
  31. 			// key
  32. 			String lockKey = "lock_" + lockName;
  33. 			// key所对应的value值
  34. 			String lockValue = UUID.randomUUID().toString();
  35. 			// 计算key的超时时间
  36. 			Integer expireTime = (int) timeout / 1000;
  37. 			// 获取锁的超时时间
  38. 			long end = System.currentTimeMillis() + acquireTimeout;
  39. 			// 循环获取锁
  40. 			while (System.currentTimeMillis() <= end) {
  41. 				// key是否存在
  42. 				if (jedis.setnx(lockKey, lockValue) == 1) {
  43. 					// 如果不存在，设置key过期时间
  44. 					jedis.expire(lockKey, expireTime);
  45. 					identifier = lockValue;
  46. 					return identifier;
  47. 				}
  48. 				// 如果设置过期时间异常，那么下一次必须设置key过期时间，否则可能出现死锁
  49. 				if (jedis.ttl(lockKey) == -1) {
  50. 					jedis.expire(lockKey, expireTime);
  51. 				}
  52. 				// 设置一个时间间隔，避免频繁尝试获取锁
  53. 				Thread.sleep(20);
  54. 			}
  55. 		} catch (Exception e) {
  56. 			e.printStackTrace();
  57. 		}
  58. 		return identifier;
  59. 	}
  60.     /**
  61. 	 * 手动释放锁
  62. 	 * 
  63. 	 * @param lockName
  64. 	 * 			锁的名称
  65. 	 */
  66. 	public boolean unlock(String lockName,String lockValue) {
  67. 		try {
  68. 			// key
  69. 			String lockKey = "lock_" + lockName;
  70. 			while(jedis.exists(lockKey)) {
  71. 				jedis.watch(lockKey);
  72. 				if(lockValue.equals(jedis.get(lockKey))) {
  73. 					Transaction transaction = jedis.multi();
  74. 					// 手动删除key,释放锁
  75. 					transaction.del(lockKey);
  76. 					List<Object> execQueue = transaction.exec();
  77. 					if(execQueue != null) {
  78. 						return true;
  79. 					}
  80. 				}
  81. 				jedis.unwatch();
  82. 				break;
  83. 			}
  84. 		} catch (Exception e) {
  85. 			e.printStackTrace();
  86. 		}
  87. 		return false;
  88. 	} 
  1. /**
  2.  * @author reyco
  3.  * @date 2021年3月18日---下午5:14:56
  4.  * 
  5.  *       <pre>
  6.  *       redis锁实现
  7.  * 
  8.  *       <pre>
  9.  */
  10. @Component
  11. public class RedisLock implements DistributedLock {
  12.  
  13. 	protected Logger logger = LoggerFactory.getLogger(this.getClass());
  14. 	
  15. 	public final static String SET_LUA_SCRIPT = "if redis.call('setnx', KEYS[1],ARGV[1]) == 1 then return redis.call('EXPIRE',KEYS[1],ARGV[2]) else return 0 end";// lua脚本，用来获取分布式锁
  16. 	
  17. 	public final static String DEL_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";// lua脚本，用来释放分布式锁
  18. 	
  19. 	public final static String RENEWAL_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('EXPIRE',KEYS[1],ARGV[2]) else return 0 end";// lua脚本，用来续约过期时间
  20. 	@Autowired
  21. 	@Qualifier("distributedLockThread")
  22. 	ExecutorService executorService;
  23. 	
  24. 	/**
  25. 	 * 当前线程锁信息
  26. 	 */
  27. 	private ThreadLocal<LockInfo> lockInfoThreadLocal = new ThreadLocal<>();
  28. 	/**
  29. 	 * 锁的key
  30. 	 */
  31. 	private static final String DISTRIBUTED_LOCK_KEY = "distributedLock:";
  32. 	/**
  33. 	 * key的名称
  34. 	 */
  35. 	private static final String DISTRIBUTED_LOCK_DEFAULT = "default";
  36. 	/**
  37. 	 * 锁的超时时间,默认3000
  38. 	 */
  39. 	private static final Integer DISTRIBUTED_LOCK_EXPIRE = 3000;
  40.  
  41. 	@Autowired
  42. 	private StringRedisTemplate redisTemplate;
  43.  
  44. 	@Override
  45. 	public void lock() {
  46. 		LockInfo lockInfo = new LockInfo(DISTRIBUTED_LOCK_KEY+DISTRIBUTED_LOCK_DEFAULT, SnowFlake.getNextId().toString(), DISTRIBUTED_LOCK_EXPIRE);
  47. 		lockInfoThreadLocal.set(lockInfo);
  48. 		List<String> keys = new ArrayList<>();
  49. 		keys.add(lockInfo.getLockKey());
  50. 		String valueExpireTime = (lockInfo.getExpireTime()/1000)+"";
  51. 		DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>(SET_LUA_SCRIPT,Long.class);
  52. 		while(redisTemplate.execute(defaultRedisScript,keys,lockInfo.getLockValue(),valueExpireTime).intValue()==0) {
  53. 			try {
  54. 				TimeUnit.MILLISECONDS.sleep(lockInfo.getExpireTime()/10);
  55. 			} catch (InterruptedException e) {
  56. 				e.printStackTrace();
  57. 			}
  58. 		}
  59. 		logger.debug("加锁成功,【key:" +lockInfo.getLockKey() + "】,【value:" + lockInfo.getLockValue() + "】,【expireTimeSecond："+lockInfo.getExpireTime()+"】");
  60. 		// 招一个看门狗
  61. 		executorService.execute(new Runnable() {
  62. 			@Override
  63. 			public void run() {
  64. 				while(lockInfo.getLockValue().equals(redisTemplate.opsForValue().get(lockInfo.getLockKey()))) {
  65. 					List<String> keys = new ArrayList<>();
  66. 					keys.add(lockInfo.getLockKey());
  67. 					DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>(RENEWAL_LUA_SCRIPT,Long.class);
  68. 					Long result = redisTemplate.execute(defaultRedisScript,keys,lockInfo.getLockValue(),valueExpireTime);
  69. 					if(result.intValue()==1) {
  70. 						logger.debug("续约成功,【key:" +lockInfo.getLockKey() + "】,【value:" + lockInfo.getLockValue() + "】,【expireTimeSecond："+lockInfo.getExpireTime()+"】");
  71. 					}
  72. 					try {
  73. 						Thread.sleep(lockInfo.getExpireTime()/3);
  74. 					} catch (InterruptedException e) {
  75. 						e.printStackTrace();
  76. 					}
  77. 				}
  78. 			}
  79. 		});
  80. 	}
  81. 	@Override
  82. 	public void unlock() {
  83. 		LockInfo lockInfo = lockInfoThreadLocal.get();
  84. 		lockInfoThreadLocal.remove();
  85. 		List<String> keys = new ArrayList<>();
  86. 		keys.add(lockInfo.getLockKey());
  87. 		DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>(DEL_LUA_SCRIPT,Long.class);
  88. 		Long integer = redisTemplate.execute(defaultRedisScript,keys,lockInfo.getLockValue());
  89. 		if(integer.intValue()==1) {
  90. 			logger.debug("解锁成功,【key:" +lockInfo.getLockKey() + "】,【value:" + lockInfo.getLockValue() + "】");
  91. 		}
  92. 	}
  93. 	
  94. 	/**
  95. 	 * 加锁
  96. 	 * 
  97. 	 * @param lockKey
  98. 	 *            锁的lockKey
  99. 	 * @param lockValue
  100. 	 *            锁的value值
  101. 	 * @param expireTime
  102. 	 *            过期时间
  103. 	 */
  104. 	@Override
  105. 	public void lock(String lockKey, String lockValue, int expireTime) {
  106. 		List<String> keys = new ArrayList<>();
  107. 		keys.add(lockKey);
  108. 		String valueExpireTime = (expireTime/1000)+"";
  109. 		DefaultRedisScript<Long> setDefaultRedisScript = new DefaultRedisScript<>(SET_LUA_SCRIPT,Long.class);
  110. 		while(redisTemplate.execute(setDefaultRedisScript,keys,lockValue,valueExpireTime).intValue()==0) {
  111. 			try {
  112. 				Thread.sleep(expireTime/10);
  113. 			} catch (InterruptedException e) {
  114. 				e.printStackTrace();
  115. 			}
  116. 		}
  117. 		logger.debug("加锁成功,【key:"+lockKey+"】,【value:"+lockValue+"】,【expireTime:"+expireTime+"】");
  118. 		//获取锁成功：招一个看门狗
  119. 		new Thread(new Runnable() {
  120. 			@Override
  121. 			public void run() {
  122. 				while(lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {
  123. 					List<String> keys = new ArrayList<>();
  124. 					keys.add(lockKey);
  125. 					DefaultRedisScript<Long> renDefaultRedisScript = new DefaultRedisScript<>(RENEWAL_LUA_SCRIPT,Long.class);
  126. 					Long result = redisTemplate.execute(renDefaultRedisScript,keys,lockValue,valueExpireTime);
  127. 					if(result.intValue()==1) {
  128. 						logger.debug("续约成功,【key:"+lockValue+ "】,【value:"+lockValue+"】,【expireTimeSecond："+valueExpireTime+"】");
  129. 					}
  130. 					try {
  131. 						Thread.sleep(expireTime/3);
  132. 					} catch (InterruptedException e) {
  133. 						e.printStackTrace();
  134. 					}
  135. 				}
  136. 			}
  137. 		}).start();
  138. 	}
  139. 	/**
  140. 	 * 解锁
  141. 	 * @param lockKey
  142. 	 * @param lockValue
  143. 	 */
  144. 	@Override
  145. 	public void unLock(String lockKey, String lockValue) {
  146. 		List<String> keys = new ArrayList<>();
  147. 		keys.add(lockKey);
  148. 		DefaultRedisScript<Long> delDefaultRedisScript = new DefaultRedisScript<>(DEL_LUA_SCRIPT,Long.class);
  149. 		Long integer = redisTemplate.execute(delDefaultRedisScript,keys,lockValue);
  150. 		if(integer.intValue()==1) {
  151. 			logger.debug("解锁成功,【key:" +lockKey + "】,【value:" + lockValue + "】");
  152. 		}
  153. 	}
  154.  
  155. 	@Override
  156. 	public void lockInterruptibly() throws InterruptedException {
  157. 	}
  158.  
  159. 	@Override
  160. 	public boolean tryLock() {
  161. 		return false;
  162. 	}
  163.  
  164. 	@Override
  165. 	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  166. 		return false;
  167. 	}
  168.  
  169. 	@Override
  170. 	public Condition newCondition() {
  171. 		return null;
  172. 	}
  173.  
  174. 	/**
  175. 	 * 锁信息
  176. 	 * 
  177. 	 * @author reyco
  178. 	 *
  179. 	 */
  180. 	static class LockInfo {
  181. 		private String lockKey;
  182. 		private String lockValue;
  183. 		private Integer expireTime;
  184.  
  185. 		public LockInfo(String lockKey, String lockValue, Integer expireTime) {
  186. 			super();
  187. 			this.lockKey = lockKey;
  188. 			this.lockValue = lockValue;
  189. 			this.expireTime = expireTime;
  190. 		}
  191.  
  192. 		public String getLockKey() {
  193. 			return lockKey;
  194. 		}
  195.  
  196. 		public void setLockKey(String lockKey) {
  197. 			this.lockKey = lockKey;
  198. 		}
  199.  
  200. 		public String getLockValue() {
  201. 			return lockValue;
  202. 		}
  203.  
  204. 		public void setLockValue(String lockValue) {
  205. 			this.lockValue = lockValue;
  206. 		}
  207.  
  208. 		public Integer getExpireTime() {
  209. 			return expireTime;
  210. 		}
  211.  
  212. 		public void setExpireTime(Integer expireTime) {
  213. 			this.expireTime = expireTime;
  214. 		}
  215. 	}
  216. }


      2，基于Zookeeper实现分布式锁；

                     2.1，使用zk的临时节点特性失效分布式锁；

                          2.1.1，采用临时节点实现分布式锁（互斥锁);

                                     将zookeeper上的一个znode看作是一把锁，所有客户端都去创建/distribute_lock 节点，最终成功创建的   那个客户端也即拥有了这把锁。用完删除掉自己创建的/distribute_lock节点就释放出锁。没有获取到锁的客 户端等待，对/distribute_lock节点watcher删除监听，如果监听到 /distribute_lock节点删除，所有客户端去争取获取锁，没有获取到锁的客户端继续监听。

                                     这种实现方式有羊群问题，会出现死锁问题.  如：监听客户端很多，所有客户端争抢锁，会出现网络阻    塞；获取锁的客户端没有及时的删除节点，所有客户端一直等待，会出现死锁问题。 

                          2.1.2，采用临时顺序节点实现分布式锁：超时机制+临时顺序节点（共享锁).

                                           /distribute_lock 已经预先存在，所有客户端在它下面创建临时顺序编号目录节点，编号最小的获得      锁，用完删除，依次方便。

                                    a，在约定的节点下面创建/distribute_lock节点；

                                    b，所有客户端在/distribute_lock节点下创建自己的临时顺序节点；

                                    c，所有客户端需要获取锁；

                                              先对所有临时顺序节点排序，然后尝试获取锁。如果获取锁，正常执行，如果没有获取到锁，监听比自己次小的节点的删除事件。

                                      此种实现方案有bug：如果中间客户端出现网络故障或宕机，就会出现多个客户端获取锁的bug。

            

  1. /**
  2.  * Zookeeper 超时机制+临时节点
  3.  * 
  4.  * @author reyco
  5.  *
  6.  */
  7. public class DistributedLock implements Lock {
  8. 	//private static final String ZK_ADDRESS = "192.168.2.107:2181,192.168.2.108:2181,192.168.2.110:2181";
  9.     private static final String ZK_ADDRESS = "192.168.241.123:2181,192.168.241.124:2181,192.168.241.125:2181";
  10. 	/**
  11. 	 * 父节点
  12. 	 */
  13. 	private String distributedLock = "/distributedLock";
  14. 	/**
  15. 	 * 子节点
  16. 	 */
  17. 	private static final String LOCKNAME = "/lock_";
  18. 	/**
  19. 	 * 当前节点
  20. 	 */
  21. 	private static ThreadLocal<String> currentPath = new ThreadLocal<>();
  22. 	/**
  23. 	 * 前一个节点
  24. 	 */
  25. 	private static ThreadLocal<String> beforePath = new ThreadLocal<>();
  26. 	/**
  27. 	 * zkc
  28. 	 */
  29. 	private ZkClient zkClient;
  30. 	/**
  31. 	 * 超时时间
  32. 	 */
  33. 	private static final int TIMIOUT = 5000;
  34.  
  35. 	/**
  36. 	 * 
  37. 	 * @param rootPath
  38. 	 */
  39. 	public DistributedLock(String rootPath) {
  40. 		this.distributedLock = rootPath;
  41. 		zkClient = new ZkClient(new ZkConnection(ZK_ADDRESS, TIMIOUT));
  42. 		if (!zkClient.exists(distributedLock)) {
  43. 			zkClient.createPersistent(distributedLock);
  44. 		}
  45. 		zkClient.setZkSerializer(new ZkSerializer() {
  46. 			@Override
  47. 			public byte[] serialize(Object data) throws ZkMarshallingError {
  48. 				return String.valueOf(data).getBytes();
  49. 			}
  50. 			@Override
  51. 			public Object deserialize(byte[] bytes) throws ZkMarshallingError {
  52. 				return new String(bytes);
  53. 			}
  54. 		});
  55. 	}
  56.  
  57. 	@Override
  58. 	public void lock() {
  59. 		// 尝试获取锁
  60. 		if (!tryLock()) {
  61. 			// 等待
  62. 			waitForLock();
  63. 			// 获取锁
  64. 			lock();
  65. 		}
  66. 	}
  67.  
  68. 	/**
  69. 	 * 
  70. 	 */
  71. 	private void waitForLock() {
  72. 		CountDownLatch countDownLatch = new CountDownLatch(1);
  73. 		// 注册监听
  74. 		IZkDataListener listener = new IZkDataListener() {
  75. 			@Override
  76. 			public void handleDataDeleted(String dataPath) throws Exception {
  77. 				System.out.println(Thread.currentThread().getName()+",监听节点被删除。。。" + dataPath);
  78. 				countDownLatch.countDown();
  79. 			}
  80. 			@Override
  81. 			public void handleDataChange(String dataPath, Object data) throws Exception {
  82. 				System.out.println("监听节点数据变化。。。" + data);
  83. 			}
  84. 		};
  85. 		zkClient.subscribeDataChanges(this.beforePath.get(), listener);
  86. 		try {
  87. 			if (this.zkClient.exists(this.beforePath.get())) {
  88. 				countDownLatch.await();
  89. 			}
  90. 		} catch (Exception e) {
  91. 			e.printStackTrace();
  92. 		}
  93. 		zkClient.unsubscribeDataChanges(this.beforePath.get(), listener);
  94. 	}
  95.  
  96. 	@Override
  97. 	public void lockInterruptibly() throws InterruptedException {
  98.  
  99. 	}
  100.  
  101. 	/**
  102. 	 * 尝试获取锁
  103. 	 */
  104. 	@Override
  105. 	public boolean tryLock() {
  106. 		// 如果存在就不创建了
  107. 		if (null == this.currentPath.get()) {
  108. 			// 创建临时顺序节点
  109. 			currentPath.set(this.zkClient.createPersistentSequential(distributedLock + LOCKNAME, "lockName"));
  110. 		}
  111. 		// 获取所有子节点
  112. 		List<String> children = this.zkClient.getChildren(distributedLock);
  113. 		// 排序children
  114. 		Collections.sort(children);
  115. 		// 判断是否最小节点
  116. 		if (currentPath.get().equals(distributedLock + "/" + children.get(0))) {
  117. 			return true;
  118. 		}
  119. 		// 获取前一个节点
  120. 		else {
  121. 			// 获取当前节点的索引号
  122. 			int currentIndex = children.indexOf(currentPath.get().substring(distributedLock.length()+1));
  123. 			beforePath.set(distributedLock + "/"+children.get(currentIndex - 1));
  124. 		}
  125. 		return false;
  126. 	}
  127.  
  128. 	@Override
  129. 	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  130. 		return false;
  131. 	}
  132.  
  133. 	@Override
  134. 	public void unlock() {
  135. 		zkClient.delete(this.currentPath.get());
  136. 	}
  137.  
  138. 	@Override
  139. 	public Condition newCondition() {
  140. 		return null;
  141. 	}
  142.  
  143. 	/**
  144. 	 * 测试
  145. 	 * @param args
  146. 	 * @throws InterruptedException
  147. 	 */
  148. 	public static void main(String[] args) throws InterruptedException {
  149. 		String path =  "/distributedLock";
  150. 		List<String> list = new ArrayList<>();
  151. 		int threadSize = 20;
  152. 		CountDownLatch countDownLatch = new  CountDownLatch(threadSize);
  153. 		ExecutorService threadPool = Executors.newFixedThreadPool(threadSize);
  154. 		for (int i = 0; i < threadSize; i++) {
  155. 			threadPool.execute(new Runnable() {
  156. 				@Override
  157. 				public void run() {
  158. 					DistributedLock distributedLock = new DistributedLock(path);
  159. 					try {
  160. 						distributedLock.lock();
  161. 						list.add("asdw ");
  162. 						countDownLatch.countDown();
  163. 					} catch (Exception e) {
  164. 						e.printStackTrace();
  165. 					}finally {
  166. 						distributedLock.unlock();
  167. 					}
  168. 				}
  169. 			});
  170. 		}
  171. 		countDownLatch.await();
  172. 		threadPool.shutdown();
  173. 		System.out.println(list.size());
  174. 	}
  175. }

                     2，解决zk临时顺序节点分布式锁的bug，采用：超时机制+永久节点 实现分布式锁（共享锁）。                                                               重构一个类似redis-ping-pong：永久顺序节点写入(客户端ip+port)，判断获取锁的节点是否能正常工作，如果能正常工作就让其正常执行业务，执行业务后删除节点；如果不能正常工作就直接删除这个节点。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/爱喝兽奶帝天荒/article/detail/776323
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号