Druid 初始化的时候有段加锁代码，接下来我们通过代码来分析对锁的理解。
public void init() throws SQLException { if (inited) { return; }
// bug fixed for dead lock, for issue #2980 DruidDriver.getInstance();
final ReentrantLock lock = this.lock; try { lock.lockInterruptibly(); } catch (InterruptedException e) { throw new SQLException("interrupt", e); }
复制代码
当多个线程同时对共享资源进行操作时，会因为线程不安全，造成数据错误。在 java 中有不同的锁机制来避免这一问题。其中 ReentrantLock 是锁的一个实现类， 它可以保证线程安全。那么它是如何保证线程安全的呢？
首先，我们来了解下 ReentrantLock 原理。
ReentrantLock 原理分析
ReentrantLock 内部实现有两种方式，公平锁和非公平锁。默认是非公平锁
public ReentrantLock() { sync = new NonfairSync();}
复制代码
也可以在构造方法中指定公平锁
public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync();}
复制代码
需要特别注意的是，它们都是直接 new 的锁对象，这两个锁对象都是继承于 Sync 这个内部类的。查看非公平锁源码如下：
/** * Sync object for non-fair locks */ static final class NonfairSync extends Sync { private static final long serialVersionUID = 7316153563782823691L;
/** * Performs lock. Try immediate barge, backing up to normal * acquire on failure. */ final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } }
复制代码
而 Sync 又是继承于 AQS 同步器的
/** * Base of synchronization control for this lock. Subclassed * into fair and nonfair versions below. Uses AQS state to * represent the number of holds on the lock. */ abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = -5179523762034025860L;
/** * Performs {@link Lock#lock}. The main reason for subclassing * is to allow fast path for nonfair version. */ abstract void lock();
/** * Performs non-fair tryLock. tryAcquire is implemented in * subclasses, but both need nonfair try for trylock method. */ //尝试获取锁 final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; } // 尝试释放锁 protected final boolean tryRelease(int releases) { int c = getState() - releases; if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; }
复制代码
因此，归根结底还是通过 AQS 的思想来实现可重入锁的。接下来分析最核心的 lock 和 unlock
public void lock() { sync.lock(); }
复制代码
可见，ReentrantLock 的 lock 方法就是调用 sync 变量的 lock()。所以说实现锁，其实就是调用 NonfairSync 的 lock()方法:
/** * Sync object for non-fair locks */ static final class NonfairSync extends Sync { private static final long serialVersionUID = 7316153563782823691L;
/** * Performs lock. Try immediate barge, backing up to normal * acquire on failure. */ final void lock() { if (compareAndSetState(0, 1))//CAS自旋 setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } }
复制代码
若自旋成功则设置当前拥有独占权的线程为当前线程
protected final void setExclusiveOwnerThread(Thread thread) { exclusiveOwnerThread = thread; }
复制代码
否则的话，执行 acquire()
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
复制代码
tryAcquire：会尝试再次通过 CAS 获取一次锁。addWaiter：将当前线程加入上面锁的双向链表（等待队列）中
private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }
复制代码
acquireQueued：通过自旋，判断当前队列节点是否可以获取锁。
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
复制代码
再看看 unlock()，释放锁的方法
public void unlock() { sync.release(1); }
复制代码
public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }
复制代码
可见，释放锁就是对 AQS 中的状态值 State 进行修改，并唤醒下一个节点。
private void unparkSuccessor(Node node) { /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0);
/* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread); }
复制代码
总结：
保证线程安全的一个有效方法就是互斥同步，互斥是实现同步的一种方法，当达到了同步，也就解决了线程安全的问题。
Doug Lea《Java 并发编程：设计原则与模式》一书中， 推荐的三个用锁的最佳实践，它们分别是：
1. 永远只在更新对象的成员变量时加锁
2. 永远只在访问可变的成员变量时加锁
3. 永远不在调用其他对象的方法时加锁
评论