基于Consul的分布式锁实现
我们在构建分布式系统的基于时候，经常需要控制对共享资源的布式互斥访问。这个时候我们就涉及到分布式锁(也称为全局锁)的锁实实现，基于目前的基于各种工具，我们已经有了大量的布式实现方式，比如：基于Redis的锁实实现、基于Zookeeper的基于实现。本文将介绍一种基于Consul 的布式Key/Value存储来实现分布式锁以及信号量的方法。
分布式锁实现
基于Consul的锁实分布式锁主要利用Key/Value存储API中的acquire和release操作来实现。acquire和release操作是基于类似Check-And-Set的云服务器操作：
acquire操作只有当锁不存在持有者时才会返回true，并且set设置的布式Value值，同时执行操作的锁实session会持有对该Key的锁，否则就返回false release操作则是基于使用指定的session来释放某个Key的锁，如果指定的布式session无效，那么会返回false，锁实否则就会set设置Value值，并返回true具体实现中主要使用了这几个Key/Value的API：
create session：https://www.consul.io/api/session.html#session_create
delete session：https://www.consul.io/api/session.html#delete-session
KV acquire/release：https://www.consul.io/api/kv.html#create-update-key
基本流程
具体实现
public class Lock { private static final String prefix = "lock/"; // 同步锁参数前缀 private ConsulClient consulClient; private String sessionName; private String sessionId = null; private String lockKey; /** * * @param consulClient * @param sessionName 同步锁的session名称 * @param lockKey 同步锁在consul的KV存储中的Key路径，会自动增加prefix前缀，方便归类查询 */ public Lock(ConsulClient consulClient, String sessionName, String lockKey) { this.consulClient = consulClient; this.sessionName = sessionName; this.lockKey = prefix + lockKey; } /** * 获取同步锁 * * @param block 是源码下载否阻塞，直到获取到锁为止 * @return */ public Boolean lock(boolean block) { if (sessionId != null) { throw new RuntimeException(sessionId + " - Already locked!"); } sessionId = createSession(sessionName); while(true) { PutParams putParams = new PutParams(); putParams.setAcquireSession(sessionId); if(consulClient.setKVValue(lockKey, "lock:" + LocalDateTime.now(), putParams).getValue()) { return true; } else if(block) { continue; } else { return false; } } } /** * 释放同步锁 * * @return */ public Boolean unlock() { PutParams putParams = new PutParams(); putParams.setReleaseSession(sessionId); boolean result = consulClient.setKVValue(lockKey, "unlock:" + LocalDateTime.now(), putParams).getValue(); consulClient.sessionDestroy(sessionId, null); return result; } /** * 创建session * @param sessionName * @return */ private String createSession(String sessionName) { NewSession newSession = new NewSession(); newSession.setName(sessionName); return consulClient.sessionCreate(newSession, null).getValue(); } }单元测试
public class TestLock { private Logger logger = Logger.getLogger(getClass()); @Test public void testLock() throws Exception { new Thread(new LockRunner(1)).start(); new Thread(new LockRunner(2)).start(); new Thread(new LockRunner(3)).start(); new Thread(new LockRunner(4)).start(); new Thread(new LockRunner(5)).start(); Thread.sleep(200000L); } class LockRunner implements Runnable { private Logger logger = Logger.getLogger(getClass()); private int flag; public LockRunner(int flag) { this.flag = flag; } @Override public void run() { Lock lock = new Lock(new ConsulClient(), "lock-session", "lock-key"); try { if (lock.lock(true)) { logger.info("Thread " + flag + " start!"); Thread.sleep(new Random().nextInt(3000L)); logger.info("Thread " + flag + " end!"); } } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } } }优化建议
本文我们实现了基于Consul的简单分布式锁，但是在实际运行时，可能会因为各种各样的意外情况导致unlock操作没有得到正确地执行，从而使得分布式锁无法释放。所以为了更完善的使用分布式锁，我们还必须实现对锁的超时清理等控制，保证即使出现了未正常解锁的情况下也能自动修复，以提升系统的健壮性。那么如何实现呢?请持续关注我的后续分解!
【本文为专栏作者“翟永超”的原创稿件，转载请通过联系作者获取授权】
戳这里，看该作者更多好文
服务器租用