多线程并发编程总结
Posted loveer
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了多线程并发编程总结相关的知识，希望对你有一定的参考价值。
多线程的优缺点
多线程的优点：
资源利用率更好，
程序响应更快。
多线程的代价：
设计复杂，
上下文切换开销大（先存储当前线程的本地的数据，程序指针等，然后载入另一个线程的本地数据，程序指针等，最后才开始执行），
增加资源消耗（每个线程需要消耗的资源）。线程的状态
new（新建）
runnnable（可运行）
running（运行）
blocked（阻塞）
waiting（等待）
time waiting （定时等待）
terminated（终止）JMM（Java内存模型）
JMM定义了线程和主内存之间的抽象关系。
主内存是共享内存区域，所有线程都可以访问，但线程对变量的操作(读取赋值等)必须在工作内存(栈空间)中进行。
线程间的通信(传值)必须通过主内存来完成。
对于一个实例对象中的成员方法而言：
如果方法中包含本地变量是基本数据类型，将直接存储在工作内存的帧栈结构中，
但倘若本地变量是引用类型，那么该变量的引用会存储在功能内存的帧栈中，而对象实例将存储在主内存(共享数据区域，堆)中。volatile写-读的内存语义
当写一个volatile变量时，JMM会把该线程对应的本地内存中的共享变量刷新到主内存。
当读一个volatile变量时，JMM会把该线程对应的本地内存置为无效。线程接下来将从主内存中读取共享变量。锁释放和获取的内存语义
当线程释放锁时，JMM会把该线程对应的本地内存中的共享变量刷新到主内存中。
当线程获取锁时，JMM会把该线程对应的本地内存置为无效。
锁内存语义的实现：
ReentrantLock的实现依赖于java同步器框架AbstractQueuedSynchronizer（本文简称之为AQS）。
AQS使用一个整型的volatile变量（命名为state）来维护同步状态，这个volatile变量是ReentrantLock内存语义实现的关键。final 域的内存语义
1.在构造函数内对一个final域的写入，与随后把这个被构造对象的引用赋值给一个引用变量，这两个操作之间不能重排序。
2.初次读一个包含final域的对象的应用，与随后初次读这个final域，这两个操作之间不能重排序JMM是如何处理并发过程中的三大特性
JMM是围绕这在并发过程中如何处理原子性、可见性和有序性这3个特性来建立的。
JMM 只能保证对单个 volatile 变量的读/写具有原子性，但类似于volatile++这种符合操作不具有原子性，
这时候就必须借助于 synchronized 和 Lock 来保证整块代码的原子性了。
除了volatile之外，java 中还有2个关键字能实现可见性，即synchronized和final（final修饰的变量，线程安全级别最高）。Concurrent 包的实现
Java的 CAS 会使用现代处理器上提供的高效机器级别原子指令，这些原子指令以原子方式对内存执行读-改-写操作，
这是在多处理器中实现同步的关键。
volatile 变量的读/写和CAS可以实现线程之间的通信。把这些特性整合在一起，就形成了整个 Concurrent 包得以实现的基石。
Concurrent 包通用化的实现模式：
首先，声明共享变量为 volatile；
然后，使用CAS的原子条件更新来实现线程之间的同步；
同时，配合以 volatile 的读/写和CAS所具有的 volatile 读和写的内存语义来实现线程之间的通信。volatile与CAS
ReentrantLock
ReentrantLock分为公平锁和非公平锁。
使用公平锁时：
加锁方法lock()的方法调用轨迹如下：
ReentrantLock : lock()
FairSync : lock()
AbstractQueuedSynchronizer : acquire(int arg)
ReentrantLock : tryAcquire(int acquires)
解锁方法unlock()的方法调用轨迹如下：
ReentrantLock : unlock()
AbstractQueuedSynchronizer : release(int arg)
Sync : tryRelease(int releases)
非公平锁的内存语义的实现（加锁）：
ReentrantLock : lock()
NonfairSync : lock()
AbstractQueuedSynchronizer : compareAndSetState(int expect, int update)
非公平锁在调用 lock 后，首先就会调用 CAS 进行一次抢锁，
如果这个时候恰巧锁没有被占用，那么直接就获取到锁返回了。
非公平锁在 CAS 失败后，和公平锁一样都会进入到 tryAcquire 方法，
在 tryAcquire 方法中，如果发现锁这个时候被释放了（state == 0），
非公平锁会直接 CAS 抢锁，但是公平锁会判断等待队列是否有线程处于等待状态，如果有则不去抢锁。AQS
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
//Node 的数据就是 thread + waitStatus + pre + next 四个属性。
static final class Node {
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter; //用于实现条件队列的单向链表
}
private transient volatile Node head; //当前持有锁的线程
private transient volatile Node tail; //新进来的线程
private volatile int state; //当前锁的状态，0代表没有被占用，大于 0 代表有线程持有当前锁
}AQS-node
结点状态 waitStatus
CANCELLED(1)：
表示当前结点已取消。
当 timeout 或被中断（响应中断的情况下），会触发变更为此状态，进入该状态后的结点将不会再变化。
SIGNAL(-1)：
表示后继结点在等待当前结点唤醒。后继结点入队时，会将前继结点的状态更新为SIGNAL。
CONDITION(-2)：
表示结点等待在Condition上，
当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
PROPAGATE(-3)：
共享模式下，前继结点不仅会唤醒其后继结点，同时也可能会唤醒后继的后继结点。
0：新结点入队时的默认状态。ReentrantLock.lock() 源码分析
public class ReentrantLock implements Lock, java.io.Serializable {
public void lock() {
sync.acquire(1); //调用下面
}
}
AbstractQueuedSynchronizer：
public final void acquire(int arg) {
if (!tryAcquire(arg) && //调用下面，尝试获取锁
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //尝试失败，挂起线程，放在等待队列
selfInterrupt();
}
//将线程包装成Node，放在阻塞队列最后
private Node addWaiter(Node mode) {
Node node = new Node(mode);
for (;;) {
Node oldTail = tail;
if (oldTail != null) {
node.setPrevRelaxed(oldTail);
//用CAS把自己设置为队尾, 如果成功后，这个节点成为阻塞队列新的尾节点
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
} else {
initializeSyncQueue(); //没有尾节点，则初始化队列
}
}
}
//此时已经进入阻塞队列
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (;;) {
final Node p = node.predecessor();
//阻塞队列不包含head节点，head一般指的是占有锁的线程，head后面的才称为阻塞队列
if (p == head && tryAcquire(arg)) { //判断当前节点是否是阻塞队列的第一个节点，是就抢一抢锁
setHead(node); //抢到锁，设置当前占有锁的节点为头节点
p.next = null; // help GC
return interrupted;
}
//没抢到，或者不是阻塞队列第一个节点，则挂起，等待被前驱节点唤醒
if (shouldParkAfterFailedAcquire(p, node))
interrupted |= parkAndCheckInterrupt();
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
ReentrantLock 在内部用了内部类 Sync 来管理锁：
abstract static class Sync extends AbstractQueuedSynchronizer {}
static final class FairSync extends Sync {
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//判断队列是否有人等待，如果没人则CAS尝试获取锁
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;
}
}ReentrantLock.unlock() 源码分析
public class ReentrantLock implements Lock, java.io.Serializable {
public void unlock() {
sync.release(1);
}
}
AbstractQueuedSynchronizer：
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
//唤醒后继节点（node为当前头节点）
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
node.compareAndSetWaitStatus(ws, 0);
//下面的代码就是唤醒后继节点，但是有可能后继节点取消了等待（waitStatus==1）
//从队尾往前找，找到waitStatus<=0的所有节点中排在最前面的
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
//找到节点，唤醒线程
LockSupport.unpark(s.thread);
}
//唤醒线程以后，被唤醒的线程将从以下代码中继续往前走，获取锁，设置为头节点，然后跳出循环
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node))
interrupted |= parkAndCheckInterrupt(); //挂起线程
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//前驱节点的 waitStatus == -1，说明前驱节点状态正常，当前线程需要挂起，直接可以返回true
if (ws == Node.SIGNAL)
return true;
//前驱节点 waitStatus > 0，说明前驱节点取消了排队。找到正常的节点作为前驱节点。
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
}
return false;
}
ReentrantLock：
abstract static class Sync extends AbstractQueuedSynchronizer {
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;
}
}Condition
每个 ReentrantLock 实例可以通过调用多次 newCondition 产生多个 ConditionObject 的实例。
每个 condition 有一个关联的条件队列：
线程 1 调用 condition1.await() 方法即可将当前线程 1 包装成 Node 后加入到 条件队列 中，
然后阻塞在这里，不继续往下执行，条件队列是一个单向链表；
调用 condition1.signal() 触发一次唤醒，此时唤醒的是队头，
会将 condition1 对应的 条件队列 的 firstWaiter（队头） 移到 阻塞队列 的队尾，等待获取锁，
获取锁后 await 方法才能返回，继续往下执行。
AbstractQueuedSynchronizer：
public class ConditionObject implements Condition, java.io.Serializable {
//条件队列的第一个节点
private transient Node firstWaiter;
//条件队列的最后一个节点
private transient Node lastWaiter;
//阻塞线程，放入条件队列，等待唤醒
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter(); //添加到 condition 的条件队列中
int savedState = fullyRelease(node); //完全释放锁
int interruptMode = 0;
//阻塞，等待进入阻塞队列，直到已经移到阻塞队列或者线程中断
while (!isOnSyncQueue(node)) {
LockSupport.park(this); //线程挂起
/**
* 有以下三种情况会让 LockSupport.park(this); 这句返回继续往下执行：
* 1. 常规路径。signal -> 转移节点到阻塞队列 -> 获取了锁（unpark）。
* 2. 线程中断。在 park 的时候，另外一个线程对这个线程进行了中断。
* 3. signal 的时候，转移以后的前驱节点取消了，或者对前驱节点的CAS操作失败了。
* 4. 假唤醒。
*/
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) //判断是否发生中断
break;
}
//等待获取锁
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
//signal 唤醒线程，转移到阻塞队列
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
//从条件队列队头往后遍历，找出第一个需要转移的 node
//因为有些线程会取消排队，但是可能还在队列中
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null); //如果 first 转移不成功，那么选择 first 后面的第一个节点进行转移
}
}
final boolean transferForSignal(Node node) {
if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
return false;
Node p = enq(node); //自旋进入阻塞队列，p 是 node 在阻塞队列的前驱节点
int ws = p.waitStatus;
//ws > 0，说明 node 在阻塞队列中的前驱节点取消了等待锁，直接唤醒 node 对应的线程。
//如果 ws <= 0, 那么 compareAndSetWaitStatus 将会被调用
//( 节点入队后，需要把前驱节点的状态设为 Node.SIGNAL(-1) )
if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
//如果前驱节点取消或者 CAS 失败，会进到这里唤醒线程，在 acquireQueued 中找到合适的前驱节点，然后挂起
LockSupport.unpark(node.thread);
return true;
}阻塞队列与条件队列
AQS 共享模式
CountDownLatch：
等待计数完成才返回（栅栏）
CyclicBarrier：
可重复使用的栅栏
打破一个栅栏：
private void breakBarrier() {
//设置状态 broken 为 true
generation.broken = true;
//重置 count 为初始值 parties
count = parties;
//唤醒所有已经在等待的线程
trip.signalAll();
}
开启新的一代（自动开启下一代。除非打破栅栏）：
//开启新的一代，当最后一个线程到达栅栏上的时候，调用这个方法来唤醒其他线程，同时初始化“下一代”
private void nextGeneration() {
//首先，需要唤醒所有的在栅栏上等待的线程
trip.signalAll();
//更新 count 的值
count = parties;
//重新生成“新一代”
generation = new Generation();
}
重置：reset()
打破栅栏，所有等待的线程会唤醒，
await 方法会通过抛出 BrokenBarrierException 异常返回，然后开启新的一代
Semaphore：
资源池（资源耗尽则进入阻塞队列）JUC 一 CyclicBarrier 与 Semaphore
CountDownLatch
CyclicBarrier
以上是关于多线程并发编程总结的主要内容，如果未能解决你的问题，请参考以下文章