Java并发实践(二)

Java并发实践(二)

2019独角兽企业重金招聘Python工程师标准>>>

并发性能与可伸缩性

Amdahl定律：程序的加速比取决于程序中可并行组件与串行组件所占的比重。

假定F是必需被串行执行的部分，在包含N个处理器的机器中，最高加速比为

Speedup<= 1/(F+(1-F)/N) 

在锁竞争时导致问题：

1.降低程序可伸缩性

2.上下文切换降低性能

使用锁分解、锁分段技术可以提高伸缩性，如ConcurrentHashMap使用了16个分段锁

显式锁Lock

ReentrantLock用法

Lock lock=new ReentrantLock();

lock.lock();

try{

    //更新对象状态

}finally{

    lock.unlock();

}

ReentrantLock无参构造器实现了非公平锁

 /*** Creates an instance of {@code ReentrantLock}.* This is equivalent to using {@code ReentrantLock(false)}.*/public ReentrantLock() {sync = new NonfairSync();}/*** Creates an instance of {@code ReentrantLock} with the* given fairness policy.** @param fair {@code true} if this lock should use a fair ordering policy*/public ReentrantLock(boolean fair) {sync = fair ? new FairSync() : new NonfairSync();}

NonfairSync与FairSync都继承自内部静态类Sync，而Sync继承自AbstractQueuedSynchronizer。

AbstractQueuedSynchronizer内部维护一个 state值，用于获得锁（释放锁）时，更新state值。

/*** The synchronization state.*/private volatile int state;  

此外AbstractQueuedSynchronizer定义了CLH队列，用于存放获取锁失败的线程（Node采用Exclusive独占模式），无论是公平还是非公平实现，更新state失败都会执行入栈等待。

/*** 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());elseacquire(1);}protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}}

NonfairSync的lock()首先判断state为0（没有其他线程占用，锁处于可用状态），直接更新state值。如果成功，则获得锁。反之，尝试调用aquire(1),首先直接尝试更新state值获取锁。如果state不是0，但当前线程已经持有锁，则state值+1，此处适用业务场景为重入锁。如果失败则调用tryAcquire()中的nonfairTryAcquire().否则，获取锁失败。

 /*** 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) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}

FairSync的lock()调用acquire(),首先判断state值如果是0(没有其他线程占用锁)，如果等待队列中没有其他线程等待，则尝试更新state值，成功，则获得锁。如果state不是0，但当前线程已经持有锁，则state值+1，此处适用业务场景为重入锁。否则，获取锁失败。

 /*** Sync object for fair locks*/static final class FairSync extends Sync {private static final long serialVersionUID = -3000897897090466540L;final void lock() {acquire(1);}/*** Fair version of tryAcquire.  Don't grant access unless* recursive call or no waiters or is first.*/protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0)throw new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}}

释放锁的方法一样（Sync中定义）

 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;}

总结：在锁没有被其他线程占用情况下(state=0),非公平锁直接尝试修改state值来获取锁；公平锁则判断当前队列是否有其他线程等待，如果没有线程等待，再去尝试更新state值获得锁。如果state！=0，但当前线程已经持有锁，则state值+1，为重入锁场景实现。如果state！=0，且其他线程持有锁，则获取锁失败。

其他方法

轮询锁  Lock();

定时锁 Lock(5000,NANOTIME);

可中断锁 lock.lockInterruptly();

ReentrantLock是独占访问的，不允许多个线程同时访问被保护对象。

ReentrantLock实现：NonfairSync如果发出请求的同时锁状态变为可用，那么这个线程将跳过队列中所有等待的线程并获得这个锁；FairSync线程按照他们发出请求的顺序来获得锁。

ReentrantReadWriteLock: 读写锁允许多个读线程并发访问被保护对象，但同时只允许一个写线程访问该对象。

只有当需要实现轮询锁、定时锁、可中断锁等内置锁无法提供的功能时，使用显示锁。

ReentrantLock，ReentrantReadWriteLock内部通过AQS实现，而AQS则通过Unsafe类的CAS算法实现state状态值以及队列头尾Node值得更新。

 /*** Setup to support compareAndSet. We need to natively implement* this here: For the sake of permitting future enhancements, we* cannot explicitly subclass AtomicInteger, which would be* efficient and useful otherwise. So, as the lesser of evils, we* natively implement using hotspot intrinsics API. And while we* are at it, we do the same for other CASable fields (which could* otherwise be done with atomic field updaters).*/private static final Unsafe unsafe = Unsafe();private static final long stateOffset;private static final long headOffset;private static final long tailOffset;private static final long waitStatusOffset;private static final long nextOffset;static {try {stateOffset = unsafe.objectFieldOffset(DeclaredField("state"));headOffset = unsafe.objectFieldOffset(DeclaredField("head"));tailOffset = unsafe.objectFieldOffset(DeclaredField("tail"));waitStatusOffset = unsafe.objectFieldOffset(DeclaredField("waitStatus"));nextOffset = unsafe.objectFieldOffset(DeclaredField("next"));} catch (Exception ex) { throw new Error(ex); }}

以上代码在AQS初始化时，获得state，头部Node，尾部Node，下一个Node等变量在内存中的位置偏移量。去修改值时，调用Unsate类的compareAndSwapObject、compareAndSwapInt方法（native方法）。cas（比较并修改）方法参数定义：compareAndSwapInt（对象实例，对象在内存中位置偏移量，对象预期值，对象更新值），只有判断对象预期值与实际值相等，才会去执行修改。

public final native boolean compareAndSwapObject(Object arg0, long arg1,Object arg3, Object arg4);public final native boolean compareAndSwapInt(Object arg0, long arg1,int arg3, int arg4);

 /*** CAS head field. Used only by enq.*/private final boolean compareAndSetHead(Node update) {return unsafepareAndSwapObject(this, headOffset, null, update);}/*** CAS tail field. Used only by enq.*/private final boolean compareAndSetTail(Node expect, Node update) {return unsafepareAndSwapObject(this, tailOffset, expect, update);}/*** CAS waitStatus field of a node.*/private static final boolean compareAndSetWaitStatus(Node node,int expect,int update) {return unsafepareAndSwapInt(node, waitStatusOffset,expect, update);}/*** CAS next field of a node.*/private static final boolean compareAndSetNext(Node node,Node expect,Node update) {return unsafepareAndSwapObject(node, nextOffset, expect, update);}

转载于: