亿级流量电商详情页系统实战-32.缓存数据生产服务中的zk分布式锁解决方案的代码实现

Dear 丶 2022-10-29 06:30 108阅读 0赞

# 1.zk分布式锁的原理 #

*  我们通过去创建zk的一个临时node，来模拟给一个商品id加锁
 *  zk会给你保证说，只会创建一个临时node，其他请求过来如果再要创建临时node，就会报错，NodeExistsException
 *  那么所以说，我们的所谓上锁，其实就是去创建某个product id对应的一个临时node
    
    (1) 如果临时node创建成功了，那么说明我们成功加锁了，此时就可以去执行对redis立面数据的操作
    
    (2) 如果临时node创建失败了，说明有人已经在拿到锁了，在操作reids中的数据，那么就不断的等待，直到自己可以获取到锁为止

# 2.基于zk client api源码实现 #

<dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.4.5</version>
    </dependency>
    
    /** * ZooKeeperSession * @author Administrator * */
    public class ZooKeeperSession { 
    	
    	private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
    	
    	private ZooKeeper zookeeper;
    
    	public ZooKeeperSession() { 
    		// 去连接zookeeper server，创建会话的时候，是异步去进行的
    		// 所以要给一个监听器，说告诉我们什么时候才是真正完成了跟zk server的连接
    		try { 
    			this.zookeeper = new ZooKeeper(
    					"192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181", 
    					50000, 
    					new ZooKeeperWatcher());
    			// 给一个状态CONNECTING，连接中
    			System.out.println(zookeeper.getState());
    			
    			try { 
    				// CountDownLatch
    				// java多线程并发同步的一个工具类
    				// 会传递进去一些数字，比如说1,2 ，3 都可以
    				// 然后await()，如果数字不是0，那么久卡住，等待
    				
    				// 其他的线程可以调用coutnDown()，减1
    				// 如果数字减到0，那么之前所有在await的线程，都会逃出阻塞的状态
    				// 继续向下运行
    				
    				connectedSemaphore.await();
    			} catch(InterruptedException e) { 
    				e.printStackTrace();
    			}
    
    			System.out.println("ZooKeeper session established......");
    		} catch (Exception e) { 
    			e.printStackTrace();
    		}
    	}
    	
    	/** * 获取分布式锁 * @param productId */
    	public void acquireDistributedLock(Long productId) { 
    		String path = "/product-lock-" + productId;
    	
    		try { 
    			zookeeper.create(path, "".getBytes(), 
    					Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    			System.out.println("success to acquire lock for product[id=" + productId + "]");  
    		} catch (Exception e) { 
    			// 如果那个商品对应的锁的node，已经存在了，就是已经被别人加锁了，那么就这里就会报错
    			// NodeExistsException
    			int count = 0;
    			while(true) { 
    				try { 
    					Thread.sleep(20); 
    					zookeeper.create(path, "".getBytes(), 
    							Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    				} catch (Exception e2) { 
    					e2.printStackTrace();
    					count++;
    					continue;
    				}
    				System.out.println("success to acquire lock for product[id=" + productId + "] after " + count + " times try......");
    				break;
    			}
    		}
    	}
    	
    	/** * 释放掉一个分布式锁 * @param productId */
    	public void releaseDistributedLock(Long productId) { 
    		String path = "/product-lock-" + productId;
    		try { 
    			zookeeper.delete(path, -1); 
    		} catch (Exception e) { 
    			e.printStackTrace();
    		}
    	}
    	
    	/** * 建立zk session的watcher * @author Administrator * */
    	private class ZooKeeperWatcher implements Watcher { 
    
    		public void process(WatchedEvent event) { 
    			System.out.println("Receive watched event: " + event.getState());
    			if(KeeperState.SyncConnected == event.getState()) { 
    				connectedSemaphore.countDown();
    			} 
    		}
    		
    	}
    	
    	/** * 封装单例的静态内部类 * @author Administrator * */
    	private static class Singleton { 
    		
    		private static ZooKeeperSession instance;
    		
    		static { 
    			instance = new ZooKeeperSession();
    		}
    		
    		public static ZooKeeperSession getInstance() { 
    			return instance;
    		}
    		
    	}
    	
    	/** * 获取单例 * @return */
    	public static ZooKeeperSession getInstance() { 
    		return Singleton.getInstance();
    	}
    	
    	/** * 初始化单例的便捷方法 */
    	public static void init() { 
    		getInstance();
    	}
    	
    }