目前几乎很多大型网站及应用都是分布式部署的，分布式场景中的数据一致性问题一直是一个比较重要的话题。分布式的CAP理论告诉我们“任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。”所以，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证“最终一致性”，只要这个最终时间是在用户可以接受的范围内即可。在很多场景中，我们为了保证数据的最终一致性，需要很多的技术方案来支持，比如分布式事务、分布式锁等。
什么是分布式锁
分布式锁一般用在分布式系统或者多个应用中，用来控制同一任务是否执行或者任务的执行顺序。在项目中，部署了多个tomcat应用，在执行定时任务时就会遇到同一任务可能执行多次的情况，我们可以借助分布式锁，保证在同一时间只有一个tomcat应用执行了定时任务。
分布式锁有那些实现方案
1.使用数据库实现分布式锁（不推荐使用）
缺点:效率低，性能差、线程出现异常时，容易出现死锁。
2.使用redis实现分布式锁
缺点:锁的失效时间难控制、容易产生死锁、非阻塞式、不可重入。
3.使用zookeeper实现分布式锁
实现相对简单、可靠性强、使用临时节点，失效时间容易控制。
4.springcloud内置实现全局锁
比较冷门，但是效率最高，实现简单，失效时间容易控制。
应用场景
分布式锁主要用于在分布式环境中保护跨进程、跨主机、跨网络的共享资源实现互斥访问，以达到保证数据的一致性。
实现原理
利用Zookeeper来实现分布式锁，主要基于其临时(或临时有序)节点和watch机制。
架构分析
实现思路和流程
1、在zookeeper指定节点（locker）下创建临时顺序节点node_n。
2、获取locker下所有子节点children。
3、对子节点按节点自增序号从小到大排序。
4、判断本节点是不是第一个子节点，若是，则获取锁；若不是，则等待。
5、使用zookeeper感知节点的功能，对本节点的上一个节点进行感知。
6、当上一个节点被删除了，zookeeper会通知该线程，该线程就结束等待，并获取锁。
7、释放锁，并删除该临时节点。
具体实现代码
下面就具体使用java和zookeeper实现分布式锁，操作zookeeper使用的是apache提供的zookeeper的包。
分布式锁类
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; /** * Lock：Java的锁接口，并实现该接口的抽象方法 * Watcher：zookeeper的节点感知接口，并实现process方法，当节点有改变时，会调用该方法 */ public class DistributedLock implements Lock, Watcher { private ZooKeeper zk = null; // 根节点 private String ROOT_LOCK = "/locker"; // 竞争的资源 private String lockName; // 等待的前一个锁 private String WAIT_LOCK; // 当前锁 private String CURRENT_LOCK; // 同步计数器 private CountDownLatch countDownLatch; private int sessionTimeout = 30000; /** * 配置分布式锁 * @param config 连接的url * @param lockName 竞争资源 */ public DistributedLock(String config, String lockName) { this.lockName = lockName; try { // 连接zookeeper zk = new ZooKeeper(config, sessionTimeout, this); Stat stat = zk.exists(ROOT_LOCK, false); if (stat == null) { // 如果根节点不存在，则创建根节点 zk.create(ROOT_LOCK, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } // 节点感知器，感知到节点变化会调用该方法 public void process(WatchedEvent event) { System.out.println("感知节点变化类型："+event.getType().name()); if (this.countDownLatch != null) { //如果同步计数器不为null，则减一 this.countDownLatch.countDown(); } } public void lock() { try { if (this.tryLock()) { System.out.println(Thread.currentThread().getName() + " " + lockName + "获得了锁"); return; } else { // 等待锁 waitForLock(WAIT_LOCK, sessionTimeout); } } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } public boolean tryLock() { try { String splitStr = "_lock_"; if (lockName.contains(splitStr)) { throw new RuntimeException("锁名有误"); } // 创建临时有序节点 CURRENT_LOCK = zk.create(ROOT_LOCK + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println(CURRENT_LOCK + " 已经创建"); // 取所有子节点 List subNodes = zk.getChildren(ROOT_LOCK, false); // 取出所有lockName的锁 List lockObjects = new ArrayList(); for (String node : subNodes) { String _node = node.split(splitStr)[0]; if (_node.equals(lockName)) { lockObjects.add(node); } } Collections.sort(lockObjects); System.out.println(Thread.currentThread().getName() + " 的锁是 " + CURRENT_LOCK); // 若当前节点为最小节点，则获取锁成功 if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) { return true; } // 若不是最小节点，则找到自己的前一个节点 String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1); WAIT_LOCK = lockObjects.get(Collections.binarySearch(lockObjects, prevNode) - 1); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } return false; } public boolean tryLock(long timeout, TimeUnit unit) { try { if (this.tryLock()) { return true; } return waitForLock(WAIT_LOCK, timeout); } catch (Exception e) { e.printStackTrace(); } return false; } // 等待锁 private boolean waitForLock(String prev, long waitTime) throws KeeperException, InterruptedException { //获取并感知上一个节点 Stat stat = zk.exists(ROOT_LOCK + "/" + prev, true); if (stat != null) { System.out.println(Thread.currentThread().getName() + "等待锁 " + ROOT_LOCK + "/" + prev); //初始化同步计数器，计数为1，当同步计数器为0，主线程才会向下执行 this.countDownLatch = new CountDownLatch(1); // 计数等待，若等到前一个节点消失，则precess中进行countDown，停止等待，获取锁 this.countDownLatch.await(waitTime, TimeUnit.MILLISECONDS); this.countDownLatch = null; System.out.println(Thread.currentThread().getName() + " 等到了锁"); } return true; } public void unlock() { try { System.out.println("释放锁 " + CURRENT_LOCK); zk.delete(CURRENT_LOCK, -1); CURRENT_LOCK = null; zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } public Condition newCondition() { return null; } public void lockInterruptibly() throws InterruptedException { this.lock(); } } |
测试类
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 | public class Test { static int n = 500; public static void secskill() { System.out.println(--n); } public static void main(String[] args) { Runnable runnable = new Runnable() { public void run() { DistributedLock lock = null; try { lock = new DistributedLock("127.0.0.1:2181", "node"); lock.lock(); secskill(); System.out.println(Thread.currentThread().getName() + "正在运行"); } finally { if (lock != null) { lock.unlock(); } } } }; for (int i = 0; i < 10; i++) { Thread t = new Thread(runnable); t.start(); } } } |
本文已通过「原本」原创作品认证，未经作者授权请勿直接转载，负责将依法追究其法律责任。
如果认为本文对您有所帮助请赞助本站
支付宝扫一扫赞助
微信钱包扫描赞助
赏