Java Juc學習筆記

Java JUC 簡介

在 Java 5.0 提供了 java.util.concurrent (簡稱
JUC )包,在此包中增加了在並發編程中很常用
的實用工具類,用於定義類似於線程的自定義子
系統,包括線程池、異步 IO 和輕量級任務框架。
提供可調的、靈活的線程池。還提供了設計用於
多線程上下文中的 Collection 實現等

 volatile 關鍵字-內存可見性

內存可見性

內存可見性(Memory Visibility)

是指當某個線程正在使用對象狀態
而另一個線程在同時修改該狀態,需要確保當一個線程修改了對象
狀態后,其他線程能夠看到發生的狀態變化。

可見性錯誤是指當讀操作與寫操作在不同的線程中執行時,我們無
法確保執行讀操作的線程能適時地看到其他線程寫入的值,有時甚
至是根本不可能的事情。

我們可以通過同步來保證對象被安全地發布。除此之外我們也可以
使用一種更加輕量級的 volatile 變量

volatile 關鍵字

Java 提供了一種稍弱的同步機制,即 volatile 變
量,用來確保將變量的更新操作通知到其他線程。
可以將 volatile 看做一個輕量級的鎖,但是又與
鎖有些不同:
對於多線程,不是一種互斥關系
不能保證變量狀態的“原子性操作”

/*
 * 一、volatile 關鍵字：當多個線程進行操作共享數據時，可以保證內存中的數據可見。
 *                       相較於 synchronized 是一種較為輕量級的同步策略。
 * 
 * 注意：
 * 1. volatile 不具備“互斥性”
 * 2. volatile 不能保證變量的“原子性”
 */
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 volatile boolean flag = false;

    @Override
    public void run() {
        
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
        }

        flag = true;
        
        System.out.println("flag=" + isFlag());

    }

    public boolean isFlag() {
        return flag;
    }

    public void setFlag(boolean flag) {
        this.flag = flag;
    }

}

 

原子變量-CAS算法 

CAS 算法

/*
 * 模擬 CAS 算法
 */
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 expectedValue = cas.get();
                    boolean b = cas.compareAndSet(expectedValue, (int)(Math.random() * 101));
                    System.out.println(b);
                }
            }).start();
        }
        
    }
    
}

class CompareAndSwap{
    private int value;
    
    //獲取內存值
    public synchronized int get(){
        return value;
    }
    
    //比較
    public synchronized int compareAndSwap(int expectedValue, int newValue){
        int oldValue = value;
        
        if(oldValue == expectedValue){
            this.value = newValue;
        }
        
        return oldValue;
    }
    
    //設置
    public synchronized boolean compareAndSet(int expectedValue, int newValue){
        return expectedValue == compareAndSwap(expectedValue, newValue);
    }
}

 

CAS (Compare-And-Swap) 是一種硬件對並發的支持,針對多處理器
操作而設計的處理器中的一種特殊指令,用於管理對共享數據的並
發訪問。
CAS 是一種無鎖的非阻塞算法的實現。
CAS 包含了 3 個操作數:
需要讀寫的內存值 V
進行比較的值 A
擬寫入的新值 B
當且僅當 V 的值等於 A 時,CAS 通過原子方式用新值 B 來更新 V 的
值,否則不會執行任何操作。

 

原子變量

import java.util.concurrent.atomic.AtomicInteger;

/*
 * 一、i++ 的原子性問題：i++ 的操作實際上分為三個步驟“讀-改-寫”
 *           int i = 10;
 *           i = i++; //10
 * 
 *           int temp = i;
 *           i = i + 1;
 *           i = temp;
 * 
 * 二、原子變量：在 java.util.concurrent.atomic 包下提供了一些原子變量。
 *         1. volatile 保證內存可見性
 *         2. CAS（Compare-And-Swap） 算法保證數據變量的原子性
 *             CAS 算法是硬件對於並發操作的支持
 *             CAS 包含了三個操作數：
 *             ①內存值  V
 *             ②預估值  A
 *             ③更新值  B
 *             當且僅當 V == A 時， V = B; 否則，不會執行任何操作。
 */
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 volatile int serialNumber = 0;
    
    private AtomicInteger serialNumber = new AtomicInteger(0);

    @Override
    public void run() {
        
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
        }
        
        System.out.println(getSerialNumber());
    }
    
    public int getSerialNumber(){
        return serialNumber.getAndIncrement();
    }
    
    
}

 

 

類的小工具包,支持在單個變量上解除鎖的線程安全編程。事實上,此包中的類可
將 volatile 值、字段和數組元素的概念擴展到那些也提供原子條件更新操作的類。
類 AtomicBoolean、AtomicInteger、AtomicLong 和 AtomicReference 的實例各自提供對
相應類型單個變量的訪問和更新。每個類也為該類型提供適當的實用工具方法。
AtomicIntegerArray、AtomicLongArray 和 AtomicReferenceArray 類進一步擴展了原子操
作,對這些類型的數組提供了支持。這些類在為其數組元素提供 volatile 訪問語義方
面也引人注目,這對於普通數組來說是不受支持的。
核心方法:boolean compareAndSet(expectedValue, updateValue)
java.util.concurrent.atomic 包下提供了一些原子操作的常用類:
AtomicBoolean 、AtomicInteger 、AtomicLong 、 AtomicReference
AtomicIntegerArray 、AtomicLongArray
AtomicMarkableReference
AtomicReferenceArray
AtomicStampedReference

 

 

ConcurrentHashMap 鎖分段機制

 

ConcurrentHashMap

Java 5.0 在 java.util.concurrent 包中提供了多種並發容器類來改進同步容器
的性能。

ConcurrentHashMap 同步容器類是Java 5 增加的一個線程安全的哈希表。對
與多線程的操作,介於 HashMap 與 Hashtable 之間。內部采用“鎖分段”
機制替代 Hashtable 的獨占鎖。進而提高性能。

 

此包還提供了設計用於多線程上下文中的 Collection 實現:
ConcurrentHashMap、ConcurrentSkipListMap、ConcurrentSkipListSet、
CopyOnWriteArrayList 和 CopyOnWriteArraySet。當期望許多線程訪問一個給
定 collection 時,ConcurrentHashMap 通常優於同步的 HashMap,
ConcurrentSkipListMap 通常優於同步的 TreeMap。當期望的讀數和遍歷遠遠
大於列表的更新數時,CopyOnWriteArrayList 優於同步的 ArrayList。

import java.util.Iterator;
import java.util.concurrent.CopyOnWriteArrayList;

/*
 * CopyOnWriteArrayList/CopyOnWriteArraySet : “寫入並復制”
 * 注意：添加操作多時，效率低，因為每次添加時都會進行復制，開銷非常的大。並發迭代操作多時可以選擇。
 */
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>());
    
    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");
        }
        
    }
    
}

 

CountDownLatch 閉鎖 

CountDownLatch
Java 5.0 在 java.util.concurrent 包中提供了多種並發容器類來改進同步容器
的性能。
CountDownLatch 一個同步輔助類,在完成一組正在其他線程中執行的操作
之前,它允許一個或多個線程一直等待。
閉鎖可以延遲線程的進度直到其到達終止狀態,閉鎖可以用來確保某些活
動直到其他活動都完成才繼續執行:
確保某個計算在其需要的所有資源都被初始化之后才繼續執行;
確保某個服務在其依賴的所有其他服務都已經啟動之后才啟動;
等待直到某個操作所有參與者都准備就緒再繼續執行。

import java.util.concurrent.CountDownLatch;

/*
 * CountDownLatch ：閉鎖，在完成某些運算是，只有其他所有線程的運算全部完成，當前運算才繼續執行
 */
public class TestCountDownLatch {

    public static void main(String[] args) {
        final CountDownLatch latch = new CountDownLatch(50);
        LatchDemo ld = new LatchDemo(latch);

        long start = System.currentTimeMillis();

        for (int i = 0; i < 50; i++) {
            new Thread(ld).start();
        }

        try {
            latch.await();
        } catch (InterruptedException e) {
        }

        long end = System.currentTimeMillis();

        System.out.println("耗費時間為：" + (end - start));
    }

}

class LatchDemo implements Runnable {

    private CountDownLatch latch;

    public LatchDemo(CountDownLatch latch) {
        this.latch = latch;
    }

    @Override
    public void run() {

        try {
            for (int i = 0; i < 50000; i++) {
                if (i % 2 == 0) {
                    System.out.println(i);
                }
            }
        } finally {
            latch.countDown();
        }

    }

}

 

 實現 Callable 接口 

Callable 接口
Java 5.0 在 java.util.concurrent 提供了一個新的創建執行
線程的方式:Callable 接口
Callable 接口類似於 Runnable,兩者都是為那些其實例可
能被另一個線程執行的類設計的。但是 Runnable 不會返
回結果,並且無法拋出經過檢查的異常。
Callable 需要依賴FutureTask ,FutureTask 也可以用作閉
鎖。

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

/*
 * 一、創建執行線程的方式三：實現 Callable 接口。 相較於實現 Runnable 接口的方式，方法可以有返回值，並且可以拋出異常。
 * 
 * 二、執行 Callable 方式，需要 FutureTask 實現類的支持，用於接收運算結果。  FutureTask 是  Future 接口的實現類
 */
public class TestCallable {
    
    public static void main(String[] args) {
        ThreadDemo td = new ThreadDemo();
        
        //1.執行 Callable 方式，需要 FutureTask 實現類的支持，用於接收運算結果。
        FutureTask<Integer> result = new FutureTask<>(td);
        
        new Thread(result).start();
        
        //2.接收線程運算后的結果
        try {
            Integer sum = result.get();  //FutureTask 可用於 閉鎖
            System.out.println(sum);
            System.out.println("------------------------------------");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

}

class ThreadDemo implements Callable<Integer>{

    @Override
    public Integer call() throws Exception {
        int sum = 0;
        
        for (int i = 0; i <= 100000; i++) {
            sum += i;
        }
        
        return sum;
    }
    
}

/*class ThreadDemo implements Runnable{

    @Override
    public void run() {
    }
    
}*/

 

 Lock 同步鎖

顯示鎖 Lock
在 Java 5.0 之前,協調共享對象的訪問時可以使用的機
制只有 synchronized 和 volatile 。Java 5.0 后增加了一些
新的機制,但並不是一種替代內置鎖的方法,而是當內
置鎖不適用時,作為一種可選擇的高級功能。
ReentrantLock 實現了 Lock 接口,並提供了與
synchronized 相同的互斥性和內存可見性。但相較於
synchronized 提供了更高的處理鎖的靈活性。

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/*
 * 一、用於解決多線程安全問題的方式：
 * 
 * synchronized:隱式鎖
 * 1. 同步代碼塊
 * 
 * 2. 同步方法
 * 
 * jdk 1.5 后：
 * 3. 同步鎖 Lock
 * 注意：是一個顯示鎖，需要通過 lock() 方法上鎖，必須通過 unlock() 方法進行釋放鎖
 */
public class TestLock {
    
    public static void main(String[] args) {
        Ticket ticket = new Ticket();
        
        new Thread(ticket, "1號窗口").start();
        new Thread(ticket, "2號窗口").start();
        new Thread(ticket, "3號窗口").start();
    }

}

class Ticket implements Runnable{
    
    private int tick = 100;
    
    private Lock lock = new ReentrantLock();

    @Override
    public void run() {
        while(true){
            
            lock.lock(); //上鎖
            
            try{
                if(tick > 0){
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                    }
                    
                    System.out.println(Thread.currentThread().getName() + " 完成售票，余票為：" + --tick);
                }
            }finally{
                lock.unlock(); //釋放鎖
            }
        }
    }
    
}

 

/*
 * 生產者和消費者案例
 */
public class TestProductorAndConsumer {

    public static void main(String[] args) {
        Clerk clerk = new Clerk();
        
        Productor pro = new Productor(clerk);
        Consumer cus = new Consumer(clerk);
        
        new Thread(pro, "生產者 A").start();
        new Thread(cus, "消費者 B").start();
        
        new Thread(pro, "生產者 C").start();
        new Thread(cus, "消費者 D").start();
    }
    
}

/*//店員
class Clerk{
    private int product = 0;
    
    //進貨
    public synchronized void get(){//循環次數：0
        while(product >= 1){//為了避免虛假喚醒問題，應該總是使用在循環中
            System.out.println("產品已滿！");
            
            try {
                this.wait();
            } catch (InterruptedException e) {
            }
            
        }
        
        System.out.println(Thread.currentThread().getName() + " : " + ++product);
        this.notifyAll();
    }
    
    //賣貨
    public synchronized void sale(){//product = 0; 循環次數：0
        while(product <= 0){
            System.out.println("缺貨！");
            
            try {
                this.wait();
            } catch (InterruptedException e) {
            }
        }
        
        System.out.println(Thread.currentThread().getName() + " : " + --product);
        this.notifyAll();
    }
}

//生產者
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++) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
            }
            
            clerk.get();
        }
    }
}

//消費者
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();
        }
    }
}*/

 

 

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/*
 * 生產者消費者案例：
 */
public class TestProductorAndConsumerForLock {

    public static void main(String[] args) {
        Clerk clerk = new Clerk();

        Productor pro = new Productor(clerk);
        Consumer con = new Consumer(clerk);

        new Thread(pro, "生產者 A").start();
        new Thread(con, "消費者 B").start();

//         new Thread(pro, "生產者 C").start();
//         new Thread(con, "消費者 D").start();
    }

}

class Clerk {
    private int product = 0;

    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    // 進貨
    public void get() {
        lock.lock();

        try {
            if (product >= 1) { // 為了避免虛假喚醒，應該總是使用在循環中。
                System.out.println("產品已滿！");

                try {
                    condition.await();
                } catch (InterruptedException e) {
                }

            }
            System.out.println(Thread.currentThread().getName() + " : "
                    + ++product);

            condition.signalAll();
        } finally {
            lock.unlock();
        }

    }

    // 賣貨
    public void sale() {
        lock.lock();

        try {
            if (product <= 0) {
                System.out.println("缺貨！");

                try {
                    condition.await();
                } catch (InterruptedException e) {
                }
            }

            System.out.println(Thread.currentThread().getName() + " : "
                    + --product);

            condition.signalAll();

        } finally {
            lock.unlock();
        }
    }
}

// 生產者
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++) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            clerk.get();
        }
    }
}

// 消費者
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();
        }
    }

}

 

Condition 控制線程通信 

Condition
Condition 接口描述了可能會與鎖有關聯的條件變量。這些變量在用
法上與使用 Object.wait 訪問的隱式監視器類似,但提供了更強大的
功能。需要特別指出的是,單個 Lock 可能與多個 Condition 對象關
聯。為了避免兼容性問題,Condition 方法的名稱與對應的 Object 版
本中的不同。
在 Condition 對象中,與 wait、notify 和 notifyAll 方法對應的分別是
await、signal 和 signalAll。
Condition 實例實質上被綁定到一個鎖上。要為特定 Lock 實例獲得
Condition 實例,請使用其 newCondition() 方法。

線程按序交替

線程按序交替
編寫一個程序,開啟 3 個線程,這三個線程的 ID 分別為
A、B、C,每個線程將自己的 ID 在屏幕上打印 10 遍,要
求輸出的結果必須按順序顯示。
如:ABCABCABC...... 依次遞歸

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/*
 * 編寫一個程序，開啟 3 個線程，這三個線程的 ID 分別為 A、B、C，每個線程將自己的 ID 在屏幕上打印 10 遍，要求輸出的結果必須按順序顯示。
 *    如：ABCABCABC…… 依次遞歸
 */
public class TestABCAlternate {
    
    public static void main(String[] args) {
        AlternateDemo ad = new AlternateDemo();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                
                for (int i = 1; i <= 20; i++) {
                    ad.loopA(i);
                }
                
            }
        }, "A").start();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                
                for (int i = 1; i <= 20; i++) {
                    ad.loopB(i);
                }
                
            }
        }, "B").start();
        
        new Thread(new Runnable() {
            @Override
            public void run() {
                
                for (int i = 1; i <= 20; i++) {
                    ad.loopC(i);
                    
                    System.out.println("-----------------------------------");
                }
                
            }
        }, "C").start();
    }

}

class AlternateDemo{
    
    private int number = 1; //當前正在執行線程的標記
    
    private Lock lock = new ReentrantLock();
    private Condition condition1 = lock.newCondition();
    private Condition condition2 = lock.newCondition();
    private Condition condition3 = lock.newCondition();
    
    /**
     * @param totalLoop : 循環第幾輪
     */
    public void loopA(int totalLoop){
        lock.lock();
        
        try {
            //1. 判斷
            if(number != 1){
                condition1.await();
            }
            
            //2. 打印
            for (int i = 1; i <= 1; i++) {
                System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
            }
            
            //3. 喚醒
            number = 2;
            condition2.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    public void loopB(int totalLoop){
        lock.lock();
        
        try {
            //1. 判斷
            if(number != 2){
                condition2.await();
            }
            
            //2. 打印
            for (int i = 1; i <= 1; i++) {
                System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
            }
            
            //3. 喚醒
            number = 3;
            condition3.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    public void loopC(int totalLoop){
        lock.lock();
        
        try {
            //1. 判斷
            if(number != 3){
                condition3.await();
            }
            
            //2. 打印
            for (int i = 1; i <= 1; i++) {
                System.out.println(Thread.currentThread().getName() + "\t" + i + "\t" + totalLoop);
            }
            
            //3. 喚醒
            number = 1;
            condition1.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
}

 

ReadWriteLock 讀寫鎖

讀-寫鎖 ReadWriteLock
ReadWriteLock 維護了一對相關的鎖,一個用於只讀操作,
另一個用於寫入操作。只要沒有 writer,讀取鎖可以由
多個 reader 線程同時保持。寫入鎖是獨占的。。
ReadWriteLock 讀取操作通常不會改變共享資源,但執行
寫入操作時,必須獨占方式來獲取鎖。對於讀取操作占
多數的數據結構。 ReadWriteLock 能提供比獨占鎖更高
的並發性。而對於只讀的數據結構,其中包含的不變性
可以完全不需要考慮加鎖操作

import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/*
 * 1. ReadWriteLock : 讀寫鎖
 * 
 * 寫寫/讀寫 需要“互斥”
 * 讀讀 不需要互斥
 * 
 */
public class TestReadWriteLock {

    public static void main(String[] args) {
        ReadWriteLockDemo rw = new ReadWriteLockDemo();
        
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                rw.set((int)(Math.random() * 101));
            }
        }, "Write:").start();
        
        
        for (int i = 0; i < 100; i++) {
            new Thread(new Runnable() {
                
                @Override
                public void run() {
                    rw.get();
                }
            }).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());
            this.number = number;
        }finally{
            lock.writeLock().unlock();
        }
    }
}

 

 線程八鎖

一個對象里面如果有多個synchronized方法,某一個時刻內,只要一個線程去調用
其中的一個synchronized方法了,其它的線程都只能等待,換句話說,某一個時刻
內,只能有唯一一個線程去訪問這些synchronized方法
鎖的是當前對象this,被鎖定后,其它的線程都不能進入到當前對象的其它的
synchronized方法
加個普通方法后發現和同步鎖無關
換成兩個對象后,不是同一把鎖了,情況立刻變化。
都換成靜態同步方法后,情況又變化
所有的非靜態同步方法用的都是同一把鎖——實例對象本身,也就是說如果一個實
例對象的非靜態同步方法獲取鎖后,該實例對象的其他非靜態同步方法必須等待獲
取鎖的方法釋放鎖后才能獲取鎖,可是別的實例對象的非靜態同步方法因為跟該實
例對象的非靜態同步方法用的是不同的鎖,所以毋須等待該實例對象已獲取鎖的非
靜態同步方法釋放鎖就可以獲取他們自己的鎖。
所有的靜態同步方法用的也是同一把鎖——類對象本身,這兩把鎖是兩個不同的對
象,所以靜態同步方法與非靜態同步方法之間是不會有競態條件的。但是一旦一個
靜態同步方法獲取鎖后,其他的靜態同步方法都必須等待該方法釋放鎖后才能獲取
鎖,而不管是同一個實例對象的靜態同步方法之間,還是不同的實例對象的靜態同
步方法之間,只要它們同一個類的實例對象!11-線程池

/*
 * 題目：判斷打印的 "one" or "two" ？
 * 
 * 1. 兩個普通同步方法，兩個線程，標准打印， 打印? //one  two
 * 2. 新增 Thread.sleep() 給 getOne() ,打印? //one  two
 * 3. 新增普通方法 getThree() , 打印? //three  one   two
 * 4. 兩個普通同步方法，兩個 Number 對象，打印?  //two  one
 * 5. 修改 getOne() 為靜態同步方法，打印?  //two   one
 * 6. 修改兩個方法均為靜態同步方法，一個 Number 對象?  //one   two
 * 7. 一個靜態同步方法，一個非靜態同步方法，兩個 Number 對象?  //two  one
 * 8. 兩個靜態同步方法，兩個 Number 對象?   //one  two
 * 
 * 線程八鎖的關鍵：
 * ①非靜態方法的鎖默認為  this,  靜態方法的鎖為 對應的 Class 實例
 * ②某一個時刻內，只能有一個線程持有鎖，無論幾個方法。
 */
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() {
//                number.getTwo();
                number2.getTwo();
            }
        }).start();
        
        /*new Thread(new Runnable() {
            @Override
            public void run() {
                number.getThree();
            }
        }).start();*/
        
    }

}

class Number{
    
    public static synchronized void getOne(){//Number.class
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
        }
        
        System.out.println("one");
    }
    
    public synchronized void getTwo(){//this
        System.out.println("two");
    }
    
    public void getThree(){
        System.out.println("three");
    }
    
}

 

 線程池 

第四種獲取線程的方法:線程池,一個 ExecutorService,它使用可能的幾個池線程之
一執行每個提交的任務,通常使用 Executors 工廠方法配置。

線程池可以解決兩個不同問題:由於減少了每個任務調用的開銷,它們通常可以在
執行大量異步任務時提供增強的性能,並且還可以提供綁定和管理資源(包括執行
任務集時使用的線程)的方法。每個 ThreadPoolExecutor 還維護着一些基本的統計數
據,如完成的任務數。
為了便於跨大量上下文使用,此類提供了很多可調整的參數和擴展鈎子 (hook)。但
是,強烈建議程序員使用較為方便的 Executors 工廠方法 :

Executors.newCachedThreadPool()(無界線程池,可以進行自動線程回收)
Executors.newFixedThreadPool(int)(固定大小線程池)
Executors.newSingleThreadExecutor()(單個后台線程)
它們均為大多數使用場景預定義了設置。12-線程調度

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/*
 * 一、線程池：提供了一個線程隊列，隊列中保存着所有等待狀態的線程。避免了創建與銷毀額外開銷，提高了響應的速度。
 * 
 * 二、線程池的體系結構：
 *     java.util.concurrent.Executor : 負責線程的使用與調度的根接口
 *         |--**ExecutorService 子接口: 線程池的主要接口
 *             |--ThreadPoolExecutor 線程池的實現類
 *             |--ScheduledExecutorService 子接口：負責線程的調度
 *                 |--ScheduledThreadPoolExecutor ：繼承 ThreadPoolExecutor， 實現 ScheduledExecutorService
 * 
 * 三、工具類 : Executors 
 * ExecutorService newFixedThreadPool() : 創建固定大小的線程池
 * ExecutorService newCachedThreadPool() : 緩存線程池，線程池的數量不固定，可以根據需求自動的更改數量。
 * ExecutorService newSingleThreadExecutor() : 創建單個線程池。線程池中只有一個線程
 * 
 * ScheduledExecutorService newScheduledThreadPool() : 創建固定大小的線程，可以延遲或定時的執行任務。
 */
public class TestThreadPool {
    
    public static void main(String[] args) throws Exception {
        //1. 創建線程池
        ExecutorService pool = Executors.newFixedThreadPool(5);
        
        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());
        }
        
        
        
        /*ThreadPoolDemo tpd = new ThreadPoolDemo();
        
        //2. 為線程池中的線程分配任務
        for (int i = 0; i < 10; i++) {
            pool.submit(tpd);
        }
        
        //3. 關閉線程池
        pool.shutdown();*/
    }
    
//    new Thread(tpd).start();
//    new Thread(tpd).start();

}

class ThreadPoolDemo implements Runnable{

    private int i = 0;
    
    @Override
    public void run() {
        while(i <= 100){
            System.out.println(Thread.currentThread().getName() + " : " + i++);
        }
    }
    
}

 

 線程調度

一個 ExecutorService,可安排在給定的延遲后運行或定
期執行的命令。

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/*
 * 一、線程池：提供了一個線程隊列，隊列中保存着所有等待狀態的線程。避免了創建與銷毀額外開銷，提高了響應的速度。
 * 
 * 二、線程池的體系結構：
 *     java.util.concurrent.Executor : 負責線程的使用與調度的根接口
 *         |--**ExecutorService 子接口: 線程池的主要接口
 *             |--ThreadPoolExecutor 線程池的實現類
 *             |--ScheduledExecutorService 子接口：負責線程的調度
 *                 |--ScheduledThreadPoolExecutor ：繼承 ThreadPoolExecutor， 實現 ScheduledExecutorService
 * 
 * 三、工具類 : Executors 
 * ExecutorService newFixedThreadPool() : 創建固定大小的線程池
 * ExecutorService newCachedThreadPool() : 緩存線程池，線程池的數量不固定，可以根據需求自動的更改數量。
 * ExecutorService newSingleThreadExecutor() : 創建單個線程池。線程池中只有一個線程
 * 
 * ScheduledExecutorService newScheduledThreadPool() : 創建固定大小的線程，可以延遲或定時的執行任務。
 */
public class TestScheduledThreadPool {

    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);//生成隨機數
                    System.out.println(Thread.currentThread().getName() + " : " + num);
                    return num;
                }
                
            }, 1, TimeUnit.SECONDS);
            
            System.out.println(result.get());
        }
        
        pool.shutdown();
    }
    
}

 

ForkJoinPool 分支/合並框架 工作竊取

Fork/Join 框架:就是在必要的情況下,將一個大任務,進行拆分(fork)成
若干個小任務(拆到不可再拆時),再將一個個的小任務運算的結果進
行 join 匯總。

  

Fork/Join 框架與線程池的區別
采用 “工作竊取”模式(work-stealing):
當執行新的任務時它可以將其拆分分成更小的任務執行,並將小任務加
到線程隊列中,然后再從一個隨機線程的隊列中偷一個並把它放在自己的隊
列中。
相對於一般的線程池實現,fork/join框架的優勢體現在對其中包含的任務
的處理方式上.在一般的線程池中,如果一個線程正在執行的任務由於某些
原因無法繼續運行,那么該線程會處於等待狀態。而在fork/join框架實現中,
如果某個子問題由於等待另外一個子問題的完成而無法繼續運行。那么處理
該子問題的線程會主動尋找其他尚未運行的子問題來執行.這種方式減少了
線程的等待時間,提高了性能。

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

import org.junit.Test;

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("耗費時間為：" + Duration.between(start, end).toMillis());//166-1996-10590
    }
    
    @Test
    public void test1(){
        Instant start = Instant.now();
        
        long sum = 0L;
        
        for (long i = 0L; i <= 50000000000L; i++) {
            sum += i;
        }
        
        System.out.println(sum);
        
        Instant end = Instant.now();
        
        System.out.println("耗費時間為：" + Duration.between(start, end).toMillis());//35-3142-15704
    }
    
    //java8 新特性
    @Test
    public void test2(){
        Instant start = Instant.now();
        
        Long sum = LongStream.rangeClosed(0L, 50000000000L)
                             .parallel()
                             .reduce(0L, Long::sum);
        
        System.out.println(sum);
        
        Instant end = Instant.now();
        
        System.out.println("耗費時間為：" + Duration.between(start, end).toMillis());//1536-8118
    }

}

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(); //進行拆分，同時壓入線程隊列
            
            ForkJoinSumCalculate right = new ForkJoinSumCalculate(middle+1, end);
            right.fork(); //
            
            return left.join() + right.join();
        }
    }
    
}