徹底搞懂 JUC
你知道的越多，不知道的就越多，業(yè)余的像一棵小草！
你來，我們一起精進(jìn)！你不來，我和你的競爭對(duì)手一起精進(jìn)！
編輯：業(yè)余草
blog.csdn.net/wangwenpeng0529
推薦：https://www.xttblog.com/?p=5271
簡介
在 Java 5.0 提供了 java.util.concurrent(簡稱 JUC )包,在此包中增加了在并發(fā)編程中很常用的工具類,用于定義類似于線程的自定義子系統(tǒng)，包括線程池,異步 IO 和輕量級(jí)任務(wù)框架。還提供了設(shè)計(jì)用于多線程上下文中的 Collection 實(shí)現(xiàn)等。
volatile 關(guān)鍵字
內(nèi)存可見性
內(nèi)存可見性（Memory Visibility）是指當(dāng)某個(gè)線程正在使用對(duì)象狀態(tài)而另一個(gè)線程在同時(shí)修改該狀態(tài)，需要確保當(dāng)一個(gè)線程修改了對(duì)象狀態(tài)后，其他線程能夠看到發(fā)生的狀態(tài)變化。
可見性錯(cuò)誤是指當(dāng)讀操作與寫操作在不同的線程中執(zhí)行時(shí)，我們無法確保執(zhí)行讀操作的線程能適時(shí)地看到其他線程寫入的值，有時(shí)甚至是根本不可能的事情。
我們可以通過同步來保證對(duì)象被安全地發(fā)布。除此之外我們也可以使用一種更加輕量級(jí)的 volatile 變量。
Java 提供了一種稍弱的同步機(jī)制，即 volatile 變量，用來確保將變量的更新操作通知到其他線程?？梢詫?volatile 看做一個(gè)輕量級(jí)的鎖，但是又與鎖有些不同：
對(duì)于多線程，不是一種互斥關(guān)系
不能保證變量狀態(tài)的“原子性操作
「問題代碼示例」
/**
* @ClassName TestVolatile
* @Description: Thread 已經(jīng)修改了flag，但是main線程還是拿到的false
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestVolatile {
public static void main(String[] args) {
ThreadDemo td = new ThreadDemo();
new Thread(td).start();
while (true) {
if (td.isFlag()) {
System.out.println("______________");
break;
}
}
}
}
class ThreadDemo implements Runnable {
private boolean flag = false;
public boolean isFlag() {
return flag;
}
public void setFlag(boolean flag) {
this.flag = flag;
}
@Override
public void run() {
try {
//增加這種出現(xiàn)問題的幾率
Thread.sleep(200);
} catch (Exception e) {
}
flag = true;
System.out.println("flag=" + isFlag());
}
}
兩個(gè)線程同時(shí)修改這一個(gè)flag，為什么main拿到的還是這種修改之前的值
「內(nèi)存分析」
「解決方法，加鎖」
public class TestVolatile {
public static void main(String[] args) {
ThreadDemo td = new ThreadDemo();
new Thread(td).start();
while (true) {
synchronized (td) {
if (td.isFlag()) {
System.out.println("______________");
break;
}
}
}
}
}
加了鎖，就可以讓 while 循環(huán)每次都從主存中去讀取數(shù)據(jù)，這樣就能讀取到 true了。但是一加鎖，每次只能有一個(gè)線程訪問，當(dāng)一個(gè)線程持有鎖時(shí)，其他的就會(huì)阻塞，效率就非常低了。不想加鎖，又要解決內(nèi)存可見性問題，那么就可以使用 volatile 關(guān)鍵字。
「volatile」
private volatile boolean flag = false;
volatile 關(guān)鍵字：當(dāng)多個(gè)線程進(jìn)行操作共享數(shù)據(jù)時(shí)，可以保證內(nèi)存中的數(shù)據(jù)可見。相較于 synchronized 是一種較為輕量級(jí)的同步策略。
注意：
volatile 不具備“互斥性”
volatile 不能保證變量的“原子性”
原子性
所謂原子性就是操作不可再細(xì)分
問題代碼
/**
* @ClassName TestAtomicDemo
* @Description:
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestAtomicDemo {
public static void main(String[] args) {
AtomicDemo ad = new AtomicDemo();
for (int i = 0; i < 10; i++) {
new Thread(ad).start();
}
}
}
class AtomicDemo implements Runnable {
private int serialNumber = 0;
@Override
public void run() {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread().getName() + ":" + getSerialNumber());
}
public int getSerialNumber() {
return serialNumber++;
}
}
看到這里，好像和上面的內(nèi)存可見性問題一樣。是不是加個(gè) volatile 關(guān)鍵字就可以了呢？其實(shí)不是的，因?yàn)榧恿?volatile，只是相當(dāng)于所有線程都是在主存中操作數(shù)據(jù)而已，但是不具備互斥性。比如兩個(gè)線程同時(shí)讀取主存中的 0，然后又同時(shí)自增，同時(shí)寫入主存，結(jié)果還是會(huì)出現(xiàn)重復(fù)數(shù)據(jù)。
import java.util.concurrent.atomic.AtomicInteger;
/**
* @ClassName TestAtomicDemo
*
* 原子變量：在 java.util.concurrent.atomic 包下提供了一些原子變量。
* 1. volatile 保證內(nèi)存可見性
* 2. CAS（Compare-And-Swap） 算法保證數(shù)據(jù)變量的原子性
* CAS 算法是硬件對(duì)于并發(fā)操作的支持
* CAS 包含了三個(gè)操作數(shù)：
* ①內(nèi)存值 V
* ②預(yù)估值 A
* ③更新值 B
* 當(dāng)且僅當(dāng) V == A 時(shí)， V = B; 否則，不會(huì)執(zhí)行任何操作。
*
* @Description:
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestAtomicDemo {
public static void main(String[] args) {
AtomicDemo ad = new AtomicDemo();
for (int i = 0; i < 10; i++) {
new Thread(ad).start();
}
}
}
class AtomicDemo implements Runnable {
private AtomicInteger serialNumber = new AtomicInteger(0);
@Override
public void run() {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
}
System.out.println(Thread.currentThread().getName() + ":" + getSerialNumber());
}
public int getSerialNumber() {
return serialNumber.getAndIncrement();
}
}
AtomicInteger 這個(gè)玩意是具有原子性的 integer，用它替換后發(fā)現(xiàn)能保證線程安全
Connected to the target VM, address: '127.0.0.1:61323', transport: 'socket'
Thread-4:1
Thread-6:4
Thread-0:3
Thread-7:9
Thread-2:2
Thread-5:6
Thread-3:5
Thread-1:0
Thread-9:7
Thread-8:8
Disconnected from the target VM, address: '127.0.0.1:61323', transport: 'socket'
CAS 算法
「解決了原子性問題，解決了內(nèi)存可見性的問題」
CAS (Compare-And-Swap) 是一種硬件對(duì)并發(fā)的支持，針對(duì)多處理器操作而設(shè)計(jì)的處理器中的一種特殊指令，用于管理對(duì)共享數(shù)據(jù)的并發(fā)訪問。CAS 是一種無鎖的非阻塞算法的實(shí)現(xiàn)。
CAS 包含了 3 個(gè)操作數(shù)：
需要讀寫的內(nèi)存值 V 進(jìn)行比較的值 A 擬寫入的新值 B
當(dāng)且僅當(dāng) V 的值等于 A 時(shí)， CAS 通過原子方式用新值 B 來更新 V 的值，否則不會(huì)執(zhí)行任何操作。
CAS比較失敗的時(shí)候不會(huì)放棄CPU，會(huì)反復(fù)執(zhí)行，直到自己修改主內(nèi)存的數(shù)據(jù)
模擬CAS算法
/**
* @ClassName TestCompareAndSwap
* @Description: cas模擬 模擬帶鎖，底層不是帶synchronized
* cas 每次修改之前，都會(huì)執(zhí)行獲取比較操作
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestCompareAndSwap {
public static void main(String[] args) {
final CompareAndSwap cas = new CompareAndSwap();
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
@Override
public void run() {
int expectValue = cas.getValue();
System.out.println(cas.compareAndSet(expectValue, (int) Math.random() * 101));
}
}).start();
}
}
}
class CompareAndSwap {
public int value;
//獲取內(nèi)存值
public synchronized int getValue() {
return value;
}
//比較并交換
public synchronized int compareAndSwap(int expectValue, int newV) {
int oldV = value;
//內(nèi)存值和預(yù)估值一致 就替換
if (oldV == expectValue) {
this.value = newV;
}
return oldV;
}
//設(shè)置 調(diào)用比較并交換 看期望值和原來的值是否一致
public synchronized boolean compareAndSet(int expectValue, int newV) {
return expectValue == compareAndSwap(expectValue, newV);
}
}
原子變量
小工具包，支持在單個(gè)變量上解除鎖的線程安全編程。事實(shí)上，此包中的類可將 volatile 值、字段和數(shù)組元素的概念擴(kuò)展到那些也提供原子條件更新操作的類。
類 AtomicBoolean、 AtomicInteger、 AtomicLong 和 AtomicReference 的實(shí)例各自提供對(duì)相應(yīng)類型單個(gè)變量的訪問和更新。每個(gè)類也為該類型提供適當(dāng)?shù)膶?shí)用工具方法。
AtomicIntegerArray、 AtomicLongArray 和 AtomicReferenceArray 類進(jìn)一步擴(kuò)展了原子操作，對(duì)這些類型的數(shù)組提供了支持。這些類在為其數(shù)組元素提供 volatile 訪問語義方面也引人注目，這對(duì)于普通數(shù)組來說是不受支持的。
核心方法：boolean compareAndSet(expectedValue, updateValue)
java.util.concurrent.atomic 包下提供了一些原子操作的常用類:
AtomicBoolean 、
AtomicInteger 、
AtomicLong 、
AtomicReference
AtomicIntegerArray 、
AtomicLongArray
AtomicMarkableReference
AtomicReferenceArray
AtomicStampedReference3-ConcurrentHashMap
鎖分段機(jī)制ConcurrentHashMap
「線程安全的hash表 每一段都是一個(gè)獨(dú)立的鎖」
Java 5.0 在 java.util.concurrent 包中提供了多種并發(fā)容器類來改進(jìn)同步容器的性能。
ConcurrentHashMap 同步容器類是Java 5 增加的一個(gè)線程安全的哈希表。對(duì)與多線程的操作，介于 HashMap 與 Hashtable 之間。內(nèi)部采用“鎖分段”機(jī)制替代 Hashtable 的獨(dú)占鎖。進(jìn)而提高性能。
此包還提供了設(shè)計(jì)用于多線程上下文中的 Collection 實(shí)現(xiàn)：ConcurrentHashMap、 ConcurrentSkipListMap、 ConcurrentSkipListSet、CopyOnWriteArrayList 和 CopyOnWriteArraySet。當(dāng)期望許多線程訪問一個(gè)給定 collection 時(shí)， ConcurrentHashMap 通常優(yōu)于同步的 HashMap，ConcurrentSkipListMap 通常優(yōu)于同步的 TreeMap。當(dāng)期望的讀數(shù)和遍歷遠(yuǎn)遠(yuǎn)大于列表的更新數(shù)時(shí)， CopyOnWriteArrayList 優(yōu)于同步的 ArrayList。
ConcurrentHashMap就是一個(gè)線程安全的hash表。我們知道HashMap是線程不安全的，Hash Table加了鎖，是線程安全的，因此它效率低。HashTable加鎖就是將整個(gè)hash表鎖起來，當(dāng)有多個(gè)線程訪問時(shí)，同一時(shí)間只能有一個(gè)線程訪問，并行變成串行，因此效率低。所以JDK1.5后提供了ConcurrentHashMap，它采用了鎖分段機(jī)制。
1.8 以后底層又換成了CAS，把鎖分段機(jī)制放棄了。CAS基本就達(dá)到了無鎖的境界
CopyOnWrite寫入并復(fù)制
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
/*
* CopyOnWriteArrayList/CopyOnWriteArraySet : “寫入并復(fù)制”
* 注意：添加操作多時(shí)，效率低，因?yàn)槊看翁砑訒r(shí)都會(huì)進(jìn)行復(fù)制，開銷非常的大。并發(fā)迭代操作多時(shí)可以選擇。
*/
public class TestCopyOnWriteArrayList {
public static void main(String[] args) {
HelloThread ht = new HelloThread();
for (int i = 0; i < 10; i++) {
new Thread(ht).start();
}
}
}
class HelloThread implements Runnable {
//private static List<String> list = Collections.synchronizedList(new ArrayList<String>());
//每次修改都會(huì)復(fù)制 添加操作多時(shí) 不適合選這個(gè)
private static CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
static {
list.add("AA");
list.add("BB");
list.add("CC");
}
@Override
public void run() {
Iterator<String> it = list.iterator();
while (it.hasNext()) {
System.out.println(it.next());
list.add("AA");//邊迭代邊添加 會(huì)出現(xiàn)并發(fā)修改異常
}
}
}
CountDownLatch 閉鎖
閉鎖，在完成某些運(yùn)算時(shí)，只有其他所有線程的運(yùn)算全部完成，當(dāng)前運(yùn)算才繼續(xù)執(zhí)行，Java 5.0 在 java.util.concurrent 包中提供了多種并發(fā)容器類來改進(jìn)同步容器的性能。
CountDownLatch 一個(gè)同步輔助類，在完成一組正在其他線程中執(zhí)行的操作之前，它允許一個(gè)或多個(gè)線程一直等待。
閉鎖可以延遲線程的進(jìn)度直到其到達(dá)終止?fàn)顟B(tài)，閉鎖可以用來確保某些活動(dòng)直到其他活動(dòng)都完成才繼續(xù)執(zhí)行：
確保某個(gè)計(jì)算在其需要的所有資源都被初始化之后才繼續(xù)執(zhí)行;
確保某個(gè)服務(wù)在其依賴的所有其他服務(wù)都已經(jīng)啟動(dòng)之后才啟動(dòng);
等待直到某個(gè)操作所有參與者都準(zhǔn)備就緒再繼續(xù)執(zhí)行。
import java.util.concurrent.CountDownLatch;
/**
* @ClassName TestCountDownLatch
* @Description: 閉鎖操作 其他線程都執(zhí)行完成后當(dāng)前線程才能繼續(xù)執(zhí)行
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestCountDownLatch {
public static void main(String[] args) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(5);
LatchDemo ld = new LatchDemo(latch);
//計(jì)算執(zhí)行時(shí)間
long start = System.currentTimeMillis();
for (int i = 0; i < 5; i++) {
new Thread(ld).start();
}
//閉鎖 等待其他線程的執(zhí)行
latch.await();
long end = System.currentTimeMillis();
System.out.println("執(zhí)行時(shí)間===============================" + (end - start));
}
}
class LatchDemo implements Runnable {
private CountDownLatch latch;
public LatchDemo(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void run() {
synchronized (this) {
try {
for (int i = 0; i < 1000; i++) {
if (i % 2 == 0) {
System.out.println(Thread.currentThread().getName() + "-------------" + i);
}
}
} finally {
//線程執(zhí)行完畢后 countdown 減一
latch.countDown();
}
}
}
}
實(shí)現(xiàn) Callable 接口
Java 5.0 在 java.util.concurrent 提供了一個(gè)新的創(chuàng)建執(zhí)行線程的方式：Callable 接口
Callable 接口類似于 Runnable，兩者都是為那些其實(shí)例可能被另一個(gè)線程執(zhí)行的類設(shè)計(jì)的。但是 Runnable 不會(huì)返回結(jié)果，并且無法拋出經(jīng)過檢查的異常。
Callable 需要依賴FutureTask ， FutureTask 也可以用作閉鎖。
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
/**
* @ClassName TestCallable
* @Description:
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestCallable {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CallableThreadDemo td = new CallableThreadDemo();
//futureTask 實(shí)現(xiàn)類的支持 用于接收運(yùn)算結(jié)果
FutureTask<Integer> result = new FutureTask<>(td);
new Thread(result).start();//線程開始運(yùn)行
Integer sum = result.get();//等待線程執(zhí)行完成后 才能獲取到結(jié)果 也可以用于閉鎖操作作為等待項(xiàng)
System.out.println("總和" + sum);
}
}
/**
* @Description 多了一個(gè)方法的返回值 并且可以拋出異常
* @Author WangWenpeng
* @Param
*/
class CallableThreadDemo implements Callable<Integer> {
@Override
public Integer call() throws Exception {
int sum = 0;
for (int i = 0; i < 100; i++) {
sum += i;
}
return sum;
}
}
//class ThreadDemo implements Runnable{
// @Override
// public void run() {
//
// }
//}
同步鎖顯示鎖 Lock
在 Java 5.0 之前，協(xié)調(diào)共享對(duì)象的訪問時(shí)可以使用的機(jī)制只有 synchronized 和 volatile 。Java 5.0 后增加了一些新的機(jī)制，但并不是一種替代內(nèi)置鎖的方法，而是當(dāng)內(nèi)置鎖不適用時(shí)，作為一種可選擇的高級(jí)功能。
ReentrantLock 實(shí)現(xiàn)了 Lock 接口，并提供了與synchronized 相同的互斥性和內(nèi)存可見性。但相較于synchronized 提供了更高的處理鎖的靈活性。
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @ClassName TestLock
* @Description: 同步鎖 更靈活的方式
* lock上鎖 unlock釋放鎖
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestLock {
public static void main(String[] args) {
Ticket ticket = new Ticket();
new Thread(ticket, "1號(hào)窗口").start();
new Thread(ticket, "2號(hào)窗口").start();
new Thread(ticket, "3號(hào)窗口").start();
}
}
class Ticket implements Runnable {
private int ticket = 100;
private Lock lock = new ReentrantLock();
@Override
public void run() {
//這樣買票沒有問題
//while (ticket > 0) {
// System.out.println(Thread.currentThread().getName() + "完成售票，余票為" + --ticket);
//}
//放大問題出現(xiàn)的記錄 出現(xiàn)了負(fù)票號(hào)
//while (true) {
// if (ticket > 0) {
// try {
// Thread.sleep(200);
// System.out.println(Thread.currentThread().getName() + "完成售票，余票為" + --ticket);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }
//}
//顯式加鎖和釋放鎖
while (true) {
lock.lock();
try {
if (ticket > 0) {
try {
Thread.sleep(200);
System.out.println(Thread.currentThread().getName() + "完成售票，余票為" + --ticket);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} finally {
lock.unlock();
}
}
}
}
lock的等待喚醒機(jī)制
/**
* @ClassName TestProductorAndConsumer
* @Description: 生產(chǎn)者消費(fèi)者模型
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestProductorAndConsumer {
public static void main(String[] args) {
Clerk clerk = new Clerk();
Productor productor = new Productor(clerk);
Consumer consumer = new Consumer(clerk);
//沒有等待喚醒機(jī)制的時(shí)候
//生產(chǎn)者一直生產(chǎn) 不考錄消費(fèi)者 可能造成數(shù)據(jù)丟失
//消費(fèi)者一直消費(fèi) 不考慮生產(chǎn)者 可能造成重復(fù)消費(fèi)
new Thread(productor, "生產(chǎn)者a").start();
new Thread(consumer, "消費(fèi)者a").start();
}
}
/**
* 店員
*/
class Clerk {
//庫存共享數(shù)據(jù) 存在安全問題
private int product = 0;
//進(jìn)貨
public synchronized void get() {
if (product >= 10) {
System.out.println("產(chǎn)品已滿，無法添加");
try {
this.wait();
} catch (InterruptedException e) {
}
} else {
this.notifyAll();
System.out.println(Thread.currentThread().getName() + "店員進(jìn)貨1個(gè)產(chǎn)品 庫存為" + ++product);
}
}
//賣貨
public synchronized void sale() {
if (product <= 0) {
System.out.println("產(chǎn)品缺貨，無法售賣");
try {
this.wait();
} catch (InterruptedException e) {
}
} else {
System.out.println(Thread.currentThread().getName() + "店員銷售1個(gè)產(chǎn)品 庫存為" + --product);
this.notifyAll();
}
}
}
/**
* 生產(chǎn)者
*/
class Productor implements Runnable {
private Clerk clerk;
public Productor(Clerk clerk) {
this.clerk = clerk;
}
@Override
public void run() {
for (int i = 0; i < 20; i++) {
clerk.get();
}
}
}
/**
* @Description 消費(fèi)者
* @Author WangWenpeng
* @Date 6:45 2020/4/27
* @Param
*/
class Consumer implements Runnable {
private Clerk clerk;
public Consumer(Clerk clerk) {
this.clerk = clerk;
}
@Override
public void run() {
for (int i = 0; i < 20; i++) {
clerk.sale();
}
}
}
lock出問題的情況
生產(chǎn)者等待，增加出問題的幾率 庫存空位改成1
if (product >= 1) {
System.out.println("產(chǎn)品已滿，無法添加");
--------------------------------------------------------------------------------------
@Override
public void run() {
for (int i = 0; i < 20; i++) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
}
clerk.get();
}
}
消費(fèi)者等于0的時(shí)候， 兩個(gè)消費(fèi)者同時(shí)生產(chǎn)，之后停住了，沒有其他線程去喚醒，導(dǎo)致停在生產(chǎn)者這里。
解決方法，去掉else，讓他能走喚醒方法。
//進(jìn)貨
public synchronized void get() {
if (product >= 1) {
System.out.println("產(chǎn)品已滿，無法添加");
try {
this.wait();
} catch (InterruptedException e) {
}
}
this.notifyAll();
System.out.println(Thread.currentThread().getName() + "店員進(jìn)貨1個(gè)產(chǎn)品 庫存為" + ++product);
}
//賣貨
public synchronized void sale() {
if (product <= 0) {
System.out.println("產(chǎn)品缺貨，無法售賣");
try {
this.wait();
} catch (InterruptedException e) {
}
}
System.out.println(Thread.currentThread().getName() + "店員銷售1個(gè)產(chǎn)品 庫存為" + --product);
this.notifyAll();
}
讓線程能走到 notifyall，可以避免停止在生產(chǎn)者這里。
虛假喚醒
增加到兩個(gè)消費(fèi)者兩個(gè)生產(chǎn)者之后，如果現(xiàn)在沒有庫存，兩個(gè)消費(fèi)者都停止在wait，然后出現(xiàn)生產(chǎn)者將庫存加一，喚醒所有消費(fèi)者，這時(shí)候就出現(xiàn)了兩個(gè)消費(fèi)者同時(shí)去消費(fèi)一個(gè)庫存，導(dǎo)致庫存變成負(fù)數(shù)，這就是虛假喚醒。
「在object類的wait方法中，虛假喚醒是可能的，因此這個(gè)wait方法應(yīng)該總被使用在循環(huán)中」
解決方法
將代碼中的 if 換為 while 循環(huán)執(zhí)行。
//進(jìn)貨
public synchronized void get() {
while (product >= 1) { //wait使用在循環(huán)中
System.out.println("產(chǎn)品已滿，無法添加");
try {
this.wait();
} catch (InterruptedException e) {
}
}
this.notifyAll();
System.out.println(Thread.currentThread().getName() + "店員進(jìn)貨1個(gè)產(chǎn)品 庫存為" + ++product);
}
//賣貨
public synchronized void sale() {
while (product <= 0) {
System.out.println("產(chǎn)品缺貨，無法售賣");
try {
this.wait();
} catch (InterruptedException e) {
}
}
System.out.println(Thread.currentThread().getName() + "店員銷售1個(gè)產(chǎn)品 庫存為" + --product);
this.notifyAll();
}
控制線程通信Condition
Condition 接口描述了可能會(huì)與鎖有關(guān)聯(lián)的條件變量。這些變量在用法上與使用 Object.wait 訪問的隱式監(jiān)視器類似，但提供了更強(qiáng)大的功能。需要特別指出的是，單個(gè) Lock 可能與多個(gè) Condition 對(duì)象關(guān)聯(lián)。為了避免兼容性問題， Condition 方法的名稱與對(duì)應(yīng)的 Object 版本中的不同。
在 Condition 對(duì)象中，與 wait、 notify 和 notifyAll 方法對(duì)應(yīng)的分別是await、 signal 和 signalAll。Condition 實(shí)例實(shí)質(zhì)上被綁定到一個(gè)鎖上。要為特定 Lock 實(shí)例獲得Condition 實(shí)例，請(qǐng)使用其 newCondition()方法。
/**
* 店員
*/
class ClerkLock {
//庫存共享數(shù)據(jù) 存在安全問題
private int product = 0;
//使用lock，去掉synchronized this.wait和lock就是兩把鎖，用lock統(tǒng)一
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
//進(jìn)貨
public void get() {
lock.lock();
try {
while (product >= 1) {
System.out.println("產(chǎn)品已滿，無法添加");
try {
condition.await();
} catch (InterruptedException e) {
}
}
condition.signalAll();
System.out.println(Thread.currentThread().getName() + "店員進(jìn)貨1個(gè)產(chǎn)品 庫存為" + ++product);
} finally {
lock.unlock();
}
}
//賣貨
public synchronized void sale() {
lock.lock();
try {
while (product <= 0) {
System.out.println("產(chǎn)品缺貨，無法售賣");
try {
condition.await();
} catch (InterruptedException e) {
}
}
System.out.println(Thread.currentThread().getName() + "店員銷售1個(gè)產(chǎn)品 庫存為" + --product);
condition.signalAll();
} finally {
}
}
}
這里店員的代碼全部處理為 condition，用他的方法實(shí)現(xiàn)線程的通信。
線程按序交替線程按序交替
編寫一個(gè)程序，開啟 3 個(gè)線程，這三個(gè)線程的 ID 分別為A、 B、 C，每個(gè)線程將自己的 ID 在屏幕上打印 10 遍，要求輸出的結(jié)果必須按順序顯示。
如：ABCABCABC…… 依次遞歸9-ReadWriteLock 讀寫鎖讀-寫鎖 ReadWriteLock
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @ClassName TestABCAlternate
* @Description: 線程交替打印
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestABCAlternate {
public static void main(String[] args) {
Alternate alternate = new Alternate();
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i <= 20; i++) {
alternate.loopA(i);
}
}
}, "A").start();
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i <= 20; i++) {
alternate.loopB(i);
}
}
}, "B").start();
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i <= 20; i++) {
alternate.loopC(i);
}
}
}, "C").start();
}
}
class Alternate {
private int number = 1;//當(dāng)前正在執(zhí)行的線程號(hào)
private Lock lock = new ReentrantLock();
private Condition condition1 = lock.newCondition();
private Condition condition2 = lock.newCondition();
private Condition condition3 = lock.newCondition();
public void loopA(int totalLoop) {
lock.lock();
try {
//1.判斷1號(hào)線程
if (number != 1) {
condition1.await();
}
//2.開始打印
for (int i = 0; i < 5; i++) {
System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
}
//3.喚醒線程2
number = 2;
condition2.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void loopB(int totalLoop) {
lock.lock();
try {
//1.判斷1號(hào)線程
if (number != 2) {
condition2.await();
}
//2.開始打印
for (int i = 0; i < 5; i++) {
System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
}
//3.喚醒線程2
number = 3;
condition3.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void loopC(int totalLoop) {
lock.lock();
try {
//1.判斷1號(hào)線程
if (number != 3) {
condition3.await();
}
//2.開始打印
for (int i = 0; i < 5; i++) {
System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
}
//3.喚醒線程2
number = 1;
condition1.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
ReadWriteLock 讀寫鎖
ReadWriteLock 維護(hù)了一對(duì)相關(guān)的鎖，一個(gè)用于只讀操作，另一個(gè)用于寫入操作。只要沒有 writer，讀取鎖可以由多個(gè) reader 線程同時(shí)保持。寫入鎖是獨(dú)占的。
ReadWriteLock 讀取操作通常不會(huì)改變共享資源，但執(zhí)行寫入操作時(shí)，必須獨(dú)占方式來獲取鎖。對(duì)于讀取操作占多數(shù)的數(shù)據(jù)結(jié)構(gòu)。ReadWriteLock 能提供比獨(dú)占鎖更高的并發(fā)性。而對(duì)于只讀的數(shù)據(jù)結(jié)構(gòu)，其中包含的不變性可以完全不需要考慮加鎖操作。
讀鎖是多個(gè)線程可以一起，寫鎖是獨(dú)占的
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* @ClassName ReadWriteLock
* @Description: 讀寫鎖 讀和寫之間不互斥 寫和寫之間互斥
* @Author: WangWenpeng
* @Version 1.0
*/
public class TestReadWriteLock {
public static void main(String[] args) {
ReadWriteLockDemo demo = new ReadWriteLockDemo();
new Thread(new Runnable() {
@Override
public void run() {
demo.set((int) (Math.random() * 101));
}
}, "writeLock").start();
for (int i = 0; i < 100; i++) {
new Thread(new Runnable() {
@Override
public void run() {
demo.get();
}
}, "readLock-" + i).start();
}
}
}
class ReadWriteLockDemo {
private int number = 0;
private ReadWriteLock lock = new ReentrantReadWriteLock();
//讀
public void get() {
lock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName() + "讀:" + number);
} finally {
lock.readLock().unlock();
}
}
//寫
public void set(int number) {
lock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName() + "寫:" + number);
this.number = number;
} finally {
lock.writeLock().unlock();
}
}
}
「線程八鎖」
一個(gè)對(duì)象里面如果有多個(gè)synchronized方法，某一個(gè)時(shí)刻內(nèi)，只要一個(gè)線程去調(diào)用其中的一個(gè)synchronized方法了，其它的線程都只能等待，換句話說，某一個(gè)時(shí)刻內(nèi)，只能有唯一一個(gè)線程去訪問這些synchronized方法
鎖的是當(dāng)前對(duì)象this，被鎖定后，其它的線程都不能進(jìn)入到當(dāng)前對(duì)象的其它的synchronized方法
加個(gè)普通方法后發(fā)現(xiàn)和同步鎖無關(guān)
換成兩個(gè)對(duì)象后，不是同一把鎖了，情況立刻變化。
都換成靜態(tài)同步方法后，情況又變化
所有的非靜態(tài)同步方法用的都是同一把鎖——實(shí)例對(duì)象本身，也就是說如果一個(gè)實(shí)例對(duì)象的非靜態(tài)同步方法獲取鎖后，該實(shí)例對(duì)象的其他非靜態(tài)同步方法必須等待獲取鎖的方法釋放鎖后才能獲取鎖，可是別的實(shí)例對(duì)象的非靜態(tài)同步方法因?yàn)楦搶?shí)例對(duì)象的非靜態(tài)同步方法用的是不同的鎖，所以毋須等待該實(shí)例對(duì)象已獲取鎖的非靜態(tài)同步方法釋放鎖就可以獲取他們自己的鎖。
所有的靜態(tài)同步方法用的也是同一把鎖——類對(duì)象本身，這兩把鎖是兩個(gè)不同的對(duì)象，所以靜態(tài)同步方法與非靜態(tài)同步方法之間是不會(huì)有競態(tài)條件的。但是一旦一個(gè)靜態(tài)同步方法獲取鎖后，其他的靜態(tài)同步方法都必須等待該方法釋放鎖后才能獲取鎖，而不管是同一個(gè)實(shí)例對(duì)象的靜態(tài)同步方法之間，還是不同的實(shí)例對(duì)象的靜態(tài)同步方法之間，只要它們同一個(gè)類的實(shí)例對(duì)象！
/**
* 1. 兩個(gè)普通同步方法，兩個(gè)線程，標(biāo)準(zhǔn)打印， 打印? //one two
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number.getTwo();
}
}).start();
}
}
class Number {
public synchronized void getOne() {
System.out.println("one");
}
public synchronized void getTwo() {
System.out.println("two");
}
}
/**
2. 新增 Thread.sleep() 給 getOne() ,打印? //one two
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number.getTwo();
}
}).start();
}
}
class Number {
public synchronized void getOne() {
try {
Thread.sleep(3000);//讓one 睡3秒
} catch (InterruptedException e) {
}
System.out.println("one");
}
public synchronized void getTwo() {
System.out.println("two");
}
}
/*
*3. 新增普通方法 getThree() , 打印? //three one two
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number.getTwo();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number.getThree();
}
}).start();
}
}
class Number {
public synchronized void getOne() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println("one");
}
public synchronized void getTwo() {
System.out.println("two");
}
//普通方法
public void getThree(){
System.out.println("three");
}
}
/*
* 4. 兩個(gè)普通同步方法，兩個(gè) Number 對(duì)象，打印? //two one
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
Number number2 = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number2.getTwo();
}
}).start();
}
}
class Number {
public synchronized void getOne() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println("one");
}
public synchronized void getTwo() {
System.out.println("two");
}
}
/*
* 5. 修改 getOne() 為靜態(tài)同步方法，打印? //two one
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}//這樣其實(shí)不能通過類的實(shí)例訪問靜態(tài)，為演示這個(gè)問題
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number.getTwo();
}
}).start();
}
}
class Number {
//靜態(tài)同步方法
public static synchronized void getOne() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println("one");
}
public synchronized void getTwo() {
System.out.println("two");
}
}
/*
* 6. 修改兩個(gè)方法均為靜態(tài)同步方法，一個(gè) Number 對(duì)象? //one two
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}//這樣其實(shí)不能通過類的實(shí)例訪問靜態(tài)，為演示這個(gè)問題
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number.getTwo();
}
}).start();
}
}
class Number {
public static synchronized void getOne() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println("one");
}
public static synchronized void getTwo() {
System.out.println("two");
}
}
/*
* 7. 一個(gè)靜態(tài)同步方法，一個(gè)非靜態(tài)同步方法，兩個(gè) Number 對(duì)象? //two one
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
Number number2 = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}//這樣其實(shí)不能通過類的實(shí)例訪問靜態(tài)，為演示這個(gè)問題
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number2.getTwo();
}
}).start();
}
}
class Number {
public static synchronized void getOne() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println("one");
}
public synchronized void getTwo() {
System.out.println("two");
}
}
/*
* 8. 兩個(gè)靜態(tài)同步方法，兩個(gè) Number 對(duì)象? //one two
*/
public class TestThread8Monitor {
public static void main(String[] args) {
Number number = new Number();
Number number2 = new Number();
new Thread(new Runnable() {
@Override
public void run() {
number.getOne();
}//這樣其實(shí)不能通過類的實(shí)例訪問靜態(tài)，為演示這個(gè)問題
}).start();
new Thread(new Runnable() {
@Override
public void run() {
number2.getTwo();
}
}).start();
}
}
class Number {
public static synchronized void getOne() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println("one");
}
public static synchronized void getTwo() {
System.out.println("two");
}
}
線程八鎖的關(guān)鍵：
非靜態(tài)方法的鎖默認(rèn)為 this, 靜態(tài)方法的鎖為 對(duì)應(yīng)的 Class 實(shí)例
某一個(gè)時(shí)刻內(nèi)，只能有一個(gè)線程持有鎖，無論幾個(gè)方法。
線程池
第四種獲取線程的方法：線程池，一個(gè) ExecutorService，它使用可能的幾個(gè)池線程之一執(zhí)行每個(gè)提交的任務(wù)，通常使用 Executors 工廠方法配置。
線程池可以解決兩個(gè)不同問題：由于減少了每個(gè)任務(wù)調(diào)用的開銷，它們通常可以在執(zhí)行大量異步任務(wù)時(shí)提供增強(qiáng)的性能，并且還可以提供綁定和管理資源（包括執(zhí)行任務(wù)集時(shí)使用的線程）的方法。每個(gè) ThreadPoolExecutor 還維護(hù)著一些基本的統(tǒng)計(jì)數(shù)據(jù)，如完成的任務(wù)數(shù)。
為了便于跨大量上下文使用，此類提供了很多可調(diào)整的參數(shù)和擴(kuò)展鉤子 (hook)。但是，強(qiáng)烈建議程序員使用較為方便的 Executors 工廠方法 ：
Executors.newCachedThreadPool()（無界線程池，可以進(jìn)行自動(dòng)線程回收）Executors.newFixedThreadPool(int)（固定大小線程池）Executors.newSingleThreadExecutor()（單個(gè)后臺(tái)線程）它們均為大多數(shù)使用場景預(yù)定義了設(shè)置。
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
/**
* @Description 一、線程池：提供了一個(gè)線程隊(duì)列，隊(duì)列中保存著所有等待狀態(tài)的線程。避免了創(chuàng)建與銷毀額外開銷，提高了響應(yīng)的速度。
* 二、線程池的體系結(jié)構(gòu)：
* java.util.concurrent.Executor : 負(fù)責(zé)線程的使用與調(diào)度的根接口
* |--**ExecutorService 子接口: 線程池的主要接口
* |--ThreadPoolExecutor 線程池的實(shí)現(xiàn)類
* |--ScheduledExecutorService 子接口：負(fù)責(zé)線程的調(diào)度
* |--ScheduledThreadPoolExecutor ：繼承 ThreadPoolExecutor， 實(shí)現(xiàn) ScheduledExecutorService
* 三、工具類 : Executors
* ExecutorService newFixedThreadPool() : 創(chuàng)建固定大小的線程池
* ExecutorService newCachedThreadPool() : 緩存線程池，線程池的數(shù)量不固定，可以根據(jù)需求自動(dòng)的更改數(shù)量。
* ExecutorService newSingleThreadExecutor() : 創(chuàng)建單個(gè)線程池。線程池中只有一個(gè)線程
* ScheduledExecutorService newScheduledThreadPool() : 創(chuàng)建固定大小的線程，可以延遲或定時(shí)的執(zhí)行任務(wù)。
* @Author WangWenpeng
* @Param
*/
public class TestThreadPool {
public static void main(String[] args) throws Exception {
//1. 創(chuàng)建線程池
ExecutorService pool = Executors.newFixedThreadPool(5);
//submit Callable方法
List<Future<Integer>> list = new ArrayList<>();
for (int i = 0; i < 10; i++) {
Future<Integer> future = pool.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
int sum = 0;
for (int i = 0; i <= 100; i++) {
sum += i;
}
return sum;
}
});
list.add(future);
}
pool.shutdown();
for (Future<Integer> future : list) {
System.out.println(future.get());
}
//submit Runnable方法
ThreadPoolDemo tpd = new ThreadPoolDemo();
//2. 為線程池中的線程分配任務(wù)
for (int i = 0; i < 10; i++) {
pool.submit(tpd);
}
//3. 關(guān)閉線程池
pool.shutdown();
}
}
class ThreadPoolDemo implements Runnable {
private int i = 0;
@Override
public void run() {
while (i <= 100) {
System.out.println(Thread.currentThread().getName() + " : " + i++);
}
}
}
線程調(diào)度ScheduledExecutorService
一個(gè) ExecutorService，可安排在給定的延遲后運(yùn)行或定期執(zhí)行的命令。
public static void main(String[] args) throws Exception {
ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);
for (int i = 0; i < 5; i++) {
Future<Integer> result = pool.schedule(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
int num = new Random().nextInt(100);//生成隨機(jī)數(shù)
System.out.println(Thread.currentThread().getName() + " : " + num);
return num;
}
}, 1, TimeUnit.SECONDS);
System.out.println(result.get());
}
pool.shutdown();
}
ForkJoinPool 分支/合并框架
就是在必要的情況下，將一個(gè)大任務(wù)，進(jìn)行拆分(fork)成若干個(gè)小任務(wù)（拆到不可再拆時(shí)），再將一個(gè)個(gè)的小任務(wù)運(yùn)算的結(jié)果進(jìn)行 join 匯總。
「JoinFork/Join 框架與線程池的區(qū)別」
采用 “工作竊取”模式（work-stealing）：
當(dāng)執(zhí)行新的任務(wù)時(shí)它可以將其拆分分成更小的任務(wù)執(zhí)行，并將小任務(wù)加到線程隊(duì)列中，然后再從一個(gè)隨機(jī)線程的隊(duì)列中偷一個(gè)并把它放在自己的隊(duì)列中。
相對(duì)于一般的線程池實(shí)現(xiàn)， fork/join 框架的優(yōu)勢體現(xiàn)在對(duì)其中包含的任務(wù)的處理方式上。在一般的線程池中， 如果一個(gè)線程正在執(zhí)行的任務(wù)由于某些原因無法繼續(xù)運(yùn)行， 那么該線程會(huì)處于等待狀態(tài)。而在fork/join框架實(shí)現(xiàn)中，如果某個(gè)子問題由于等待另外一個(gè)子問題的完成而無法繼續(xù)運(yùn)行。那么處理該子問題的線程會(huì)主動(dòng)尋找其他尚未運(yùn)行的子問題來執(zhí)行.這種方式減少了線程的等待時(shí)間， 提高了性能。
public class TestForkJoinPool {
public static void main(String[] args) {
Instant start = Instant.now();
ForkJoinPool pool = new ForkJoinPool();
ForkJoinTask<Long> task = new ForkJoinSumCalculate(0L, 50000000000L);
Long sum = pool.invoke(task);
System.out.println(sum);
Instant end = Instant.now();
System.out.println("耗費(fèi)時(shí)間為：" + Duration.between(start, end).toMillis());//166-1996-10590
}
}
class ForkJoinSumCalculate extends RecursiveTask<Long> {
private static final long serialVersionUID = -259195479995561737L;
private long start;
private long end;
private static final long THURSHOLD = 10000L; //臨界值
public ForkJoinSumCalculate(long start, long end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
long length = end - start;
if (length <= THURSHOLD) {
long sum = 0L;
for (long i = start; i <= end; i++) {
sum += i;
}
return sum;
} else {
long middle = (start + end) / 2;
ForkJoinSumCalculate left = new ForkJoinSumCalculate(start, middle);
left.fork(); //進(jìn)行拆分，同時(shí)壓入線程隊(duì)列
ForkJoinSumCalculate right = new ForkJoinSumCalculate(middle + 1, end);
right.fork();
return left.join() + right.join();
}
}
}