1. 程式人生 > 實用技巧 >091_多執行緒(二)

091_多執行緒(二)

目錄
https://www.bilibili.com/video/BV1V4411p7EF/

執行緒同步機制

  1. 併發:同一個物件被多個執行緒同時操作。
  2. 處理多執行緒問題時,多個執行緒訪問同一個物件,並且某些執行緒還想修改這個物件,這時候我們就需要執行緒同步。
  3. 執行緒同步其實就是一種等待機制,多個需要同時訪問此物件的執行緒進入這個物件的等待池形成佇列,等待前面的執行緒使用完畢,下一個執行緒再使用。
  4. 由於同一程序的多個執行緒共享同一塊儲存空間,在帶來方便的同時,也帶來了訪問衝突問題,為了保證資料在方法中被訪問時的正確性,在訪問時加入 鎖機制synchronized,當一個執行緒獲得物件的排它鎖,獨佔資源,其他執行緒必須等待,使用後釋放鎖即可。存在以下問題:
    1. 一個執行緒持有鎖會導致其他所有需要此鎖的執行緒掛起。
    2. 在多執行緒競爭下,加鎖,釋放鎖會導致比較多的上下文切換和排程延時,引起效能問題。
    3. 如果一個優先順序高的執行緒等待一個優先順序低的執行緒釋放鎖,會導致優先順序倒置,引起效能問題。

不安全案例

package com.qing.sync;

/**
 * 不安全的買票
 * 執行緒不安全,有負數
 */
public class UnsafeBuyTicket {
    public static void main(String[] args) {
        BuyTicket buyTicket = new BuyTicket();

        new Thread(buyTicket, "楊康").start();
        new Thread(buyTicket, "郭靖").start();
        new Thread(buyTicket, "黃蓉").start();
    }
}

class BuyTicket implements Runnable {
    //票
    private int ticketNums = 10;
    @Override
    public void run() {
        //買票
        while (ticketNums > 0) {
            buy();
        }
    }
    private void buy() {
        //判斷是否有票
        if (ticketNums <= 0) {
            return;
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "買到第" + ticketNums-- + "票");
    }
}
楊康買到第10票
郭靖買到第9票
黃蓉買到第8票
楊康買到第7票
黃蓉買到第6票
郭靖買到第5票
楊康買到第4票
郭靖買到第3票
黃蓉買到第2票
黃蓉買到第1票
楊康買到第0票
郭靖買到第-1票
package com.qing.sync;

import java.util.ArrayList;
import java.util.List;

/**
 * 執行緒不安全的集合
 */
public class UnsafeList {
    public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            new Thread(()->{
                list.add(Thread.currentThread().getName());
            }).start();
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(list.size());
    }
}
9996

同步方法和同步塊

  1. 由於我們可以通過private關鍵字來保證資料物件只能被方法訪問,所以我們只需要針對方法提出一套機制,這套機制就是synchronized關鍵字,它包括兩種用法:
    1. synchronized方法。
    2. synchronized塊。
  2. synchronized方法控制對“物件”的訪問,每個物件對應一把鎖,每個synchronized方法都必須獲得呼叫該方法的物件的鎖才能執行,否則執行緒會阻塞,方法一旦執行,就獨佔該鎖,直到該方法返回才釋放鎖,後面被阻塞的執行緒才能獲得這個鎖,繼續執行。
  3. synchronized方法的缺陷:若將一個大的方法申明為synchronized將會影響效率。
  4. 同步塊:synchronized(obj){}
  5. obj稱之為同步監視器。
    1. obj可以是任何物件,但是推薦使用共享資源作為同步監視器。
    2. 同步方法中無需指定同步監視器,因為同步方法的同步監視器就是this,就是這個物件本身,或者是class。
  6. 同步監視器的執行過程:
    1. 第一個執行緒訪問,鎖定同步監視器,執行其中程式碼。
    2. 第二個執行緒訪問,發現同步監視器被鎖定,無法訪問。
    3. 第一個執行緒訪問完畢,解鎖同步監視器。
    4. 第二個執行緒訪問,發現同步監視器沒有鎖,然後鎖定並訪問。
  7. 同步塊鎖的物件是變化的量。
    private synchronized void buy() {
        //判斷是否有票
        if (ticketNums <= 0) {
            return;
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "買到第" + ticketNums-- + "票");
    }
package com.qing.sync;

import java.util.ArrayList;
import java.util.List;

/**
 * 執行緒不安全的集合
 */
public class UnsafeList {
    public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            new Thread(()->{
                synchronized (list) {
                    list.add(Thread.currentThread().getName());
                }
            }).start();
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(list.size());
    }
}
10000

JUC併發安全 CopyOnWriteArrayList

package com.qing.sync;

import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 測試JUC安全型別的集合
 */
public class TestJuc {
    public static void main(String[] args) {
        CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
        for (int i = 0; i < 30000; i++) {
            new Thread(()->{
                list.add(Thread.currentThread().getName());
            }).start();
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(list.size());
    }
}
30000

死鎖

  1. 多個執行緒各自佔有一些共享資源,並且互相等待其他執行緒佔有的資源才能執行,而導致兩個或者多個執行緒都在等待對方釋放資源,都停止執行的情形。
  2. 某一個同步塊同時擁有“兩個以上物件的鎖”時,就可能發生“死鎖”的問題。
package com.qing.sync;

/**
 * 死鎖:多個執行緒互相持有對方需要的資源,然後形成僵持。
 */
public class DeadLock {
    public static void main(String[] args) {
        Makeup t1 = new Makeup(0,"灰姑娘");
        Makeup t2 = new Makeup(1,"白雪公主");
        t1.start();
        t2.start();
    }
}

//口紅
class Lipstick {

}
//鏡子
class Mirror {

}

class Makeup extends Thread {
    //需要的資源只有一份,用static來保證只有一份
    static Lipstick lipstick = new Lipstick();
    static Mirror mirror = new Mirror();

    int choice;//選擇
    String girlName;//用化妝品的人

    public Makeup(int choice,String girlName) {
        this.choice = choice;
        this.girlName = girlName;
    }

    @Override
    public void run() {
        //化妝
        try {
            makeup();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //化妝,互相持有對方的鎖,就是需要拿到對方的資源
    private void makeup() throws InterruptedException {
        if (choice == 0) {
            synchronized (lipstick) {//獲得口紅的鎖
                System.out.println(this.girlName + "獲得口紅的鎖");
                Thread.sleep(1000);
                synchronized (mirror) {//一秒鐘後獲得鏡子的鎖
                    System.out.println(this.girlName + "獲得鏡子的鎖");
                }
            }
        } else {
            synchronized (mirror) {//獲得鏡子的鎖
                System.out.println(this.girlName + "獲得鏡子的鎖");
                Thread.sleep(1000);
                synchronized (lipstick) {//一秒鐘後獲得口紅的鎖
                    System.out.println(this.girlName + "獲得口紅的鎖");
                }
            }
        }
    }
}
灰姑娘獲得口紅的鎖
白雪公主獲得鏡子的鎖

死鎖避免方法

  1. 產生死鎖的四個必要條件:
    1. 互斥條件:一個資源每次只能被一個執行緒使用。
    2. 請求與保持條件:一個程序因請求資源而阻塞時,對已獲得的資源保持不放。
    3. 不剝奪條件:程序已獲得的資源,在未使用完之前,不能強行剝奪。
    4. 迴圈等待條件:若干程序之間形成一種頭尾相接的迴圈等待資源關係。
  2. 上面列出了死鎖的四個必要條件,只要破除其中的任意一個或多個條件就可以避免死鎖發生。

Lock(鎖)

  1. 從JDK5.0開始,Java提供了更強大的執行緒同步機制——通過顯式定義同步鎖物件來實現同步。同步鎖使用Lock物件充當。
  2. java.util.concurrent.locks.Lock介面是控制多個執行緒對共享資源進行訪問的工具。鎖提供了對共享資源的獨佔訪問,每次只能有一個執行緒對Lock物件加鎖,執行緒開始訪問共享資源之前應先獲得Lock物件。
  3. ReentrantLock類實現了Lock,它擁有與synchronized相同的併發性和記憶體語義,在實現執行緒安全的控制中,比較常用的是ReentrantLock,可以顯示加鎖、釋放鎖。
//定義Lock鎖
private final ReentrantLock lock = new ReentrantLock();

public void test() {
    try {
        lock.lock();//加鎖
        //保證執行緒安全的程式碼        
    } finally {
        lock.unlock();//解鎖
    }
}
package com.qing.sync;

import java.util.concurrent.locks.ReentrantLock;

/**
 * 測試Lock鎖
 */
public class TestLock {
    public static void main(String[] args) {
        Lock2 lock2 = new Lock2();

        new Thread(lock2).start();
        new Thread(lock2).start();
        new Thread(lock2).start();
    }
}

class Lock2 implements Runnable {

    int ticketNums = 10;

    //定義Lock鎖
    private final ReentrantLock lock = new ReentrantLock();

    @Override
    public void run() {
        while (true) {
            try {
                lock.lock();//加鎖
                if (ticketNums > 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(ticketNums--);
                } else {
                    break;
                }
            } finally {
                lock.unlock();//解鎖
            }
        }
    }
}
10
9
8
7
6
5
4
3
2
1

synchronized與Lock的對比

  1. Lock是顯式鎖(手動開啟和關閉鎖,別忘記關閉鎖);synchronized是隱式鎖,出了作用域自動釋放。
  2. Lock只有程式碼塊鎖,synchronized有程式碼塊鎖和方法鎖。
  3. 使用Lock鎖,JVM將花費較少的時間來排程執行緒,效能更好,並且具有更好的擴充套件性(提供更多的子類)。
  4. 優先使用順序:Lock>同步程式碼塊(已經進入了方法體,分配了相應資源)>同步方法(在方法體之外)。

執行緒協作:生產者消費者模式

執行緒通訊

  1. Java提供了幾個方法解決執行緒之間的通訊問題。

  1. 幾個方法均是Object類的方法,都只能在同步方法或者同步程式碼塊中使用,否則會丟擲異常IllegalMonitorStateException。

管程法

  1. 生產者:負責生產資料的模組(可能是方法,物件,執行緒,程序)。
  2. 消費者:負責處理資料的模組(可能是方法,物件,執行緒,程序)。
  3. 緩衝區:消費者不能直接使用生產者的資料,他們之間有個“緩衝區”。
  4. 生產者將生產的資料放入緩衝區,消費者從緩衝區拿出資料。

package com.qing.sync;

/**
 * 測試管程法:生產者消費者模型-->利用緩衝區解決
 * 生產者,消費者,產品,緩衝區
 */
public class TestPC {
    public static void main(String[] args) {
        SynContainer container = new SynContainer();

        new Producer(container).start();
        new Consumer(container).start();
    }
}

//生產者
class Producer extends Thread {
    SynContainer container;

    public Producer(SynContainer container) {
        this.container = container;
    }
    //生產
    @Override
    public void run() {
        for (int i = 1; i < 20; i++) {
            container.push(new Chicken(i));
            System.out.println("生產了" + i + "號雞");
        }
    }
}
//消費者
class Consumer extends Thread {
    SynContainer container;

    public Consumer(SynContainer container) {
        this.container = container;
    }
    //消費
    @Override
    public void run() {
        for (int i = 1; i < 20; i++) {
            System.out.println("消費了" + container.pop().id + "號雞");
        }
    }
}
//產品
class Chicken {
    int id;//產品編號

    public Chicken(int id) {
        this.id = id;
    }
}
//緩衝區
class SynContainer {
    //需要一個容器大小
    Chicken[] chickens = new Chicken[5];
    //容器計數器
    int count = 0;

    //生產者放入產品
    public synchronized void push(Chicken chicken) {
        //如果容器滿了,就需要等待消費者消費
        if (count == chickens.length) {
            //生產者等待
            try {
                System.out.println("push-->wait");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //如果容器沒有滿,就放入產品
        chickens[count] = chicken;
        count++;
        //通知消費者消費
        System.out.println("push-->notifyAll");
        this.notifyAll();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    //消費者消費產品
    public synchronized Chicken pop() {
        //如果容器空了,就需要等待生產者放入產品
        if (count == 0) {
            //消費者等待
            try {
                System.out.println("pop-->wait");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //如果容器沒有空,就消費產品
        count--;
        Chicken chicken = chickens[count];
        //通知生產者生產
        System.out.println("pop-->notifyAll");
        this.notifyAll();
        return chicken;
    }
}
pop-->wait
push-->notifyAll
生產了1號雞
push-->notifyAll
pop-->notifyAll
生產了2號雞
push-->notifyAll
消費了2號雞
pop-->notifyAll
生產了3號雞
消費了3號雞
push-->notifyAll
pop-->notifyAll
生產了4號雞
消費了4號雞
push-->notifyAll
生產了5號雞
pop-->notifyAll
消費了5號雞
push-->notifyAll
pop-->notifyAll
生產了6號雞
消費了6號雞
push-->notifyAll
pop-->notifyAll
消費了7號雞
pop-->notifyAll
生產了7號雞
消費了1號雞
push-->notifyAll
pop-->notifyAll
消費了8號雞
pop-->wait
生產了8號雞
push-->notifyAll
pop-->notifyAll
生產了9號雞
消費了9號雞
push-->notifyAll
pop-->notifyAll
生產了10號雞
push-->notifyAll
消費了10號雞
pop-->notifyAll
消費了11號雞
pop-->wait
生產了11號雞
push-->notifyAll
pop-->notifyAll
生產了12號雞
消費了12號雞
push-->notifyAll
pop-->notifyAll
生產了13號雞
消費了13號雞
push-->notifyAll
pop-->notifyAll
生產了14號雞
push-->notifyAll
消費了14號雞
pop-->notifyAll
消費了15號雞
pop-->wait
生產了15號雞
push-->notifyAll
pop-->notifyAll
生產了16號雞
消費了16號雞
push-->notifyAll
pop-->notifyAll
消費了17號雞
pop-->wait
生產了17號雞
push-->notifyAll
pop-->notifyAll
生產了18號雞
push-->notifyAll
消費了18號雞
生產了19號雞
pop-->notifyAll
消費了19號雞

訊號燈法

package com.qing.sync;

/**
 * 測試訊號燈法:生產者消費者模型--》標誌位解決
 */
public class TestPC2 {
    public static void main(String[] args) {
        TV tv = new TV();

        new Player(tv).start();
        new Watcher(tv).start();
    }
}

//生產者-->演員
class Player extends Thread {
    TV tv;

    public Player(TV tv) {
        this.tv = tv;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            if (i%2 == 0) {
                this.tv.play("天上");
            } else {
                this.tv.play("人間");
            }
        }
    }
}
//消費者-->觀眾
class Watcher extends Thread {
    TV tv;

    public Watcher(TV tv) {
        this.tv = tv;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            this.tv.watch();
        }
    }
}
//產品-->節目
class TV {
    //演員表演,觀眾等待 T
    //觀眾觀看,演員等待 F
    String voice;//表演的節目
    boolean flag = true;

    //表演
    public synchronized void play(String voice) {
        if (!flag) {
            try {
                System.out.println("play-->wait");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("演員表演:" + voice);
        //通知觀眾觀看
        System.out.println("play-->notifyAll");
        this.notifyAll();
        this.voice = voice;
        this.flag = !this.flag;
    }
    //觀看
    public synchronized void watch() {
        if (flag) {
            try {
                System.out.println("watch-->wait");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("觀眾觀看:" + voice);
        //通知演員表演
        System.out.println("watch-->notifyAll");
        this.notifyAll();
        this.voice = voice;
        this.flag = !this.flag;
    }
}
演員表演:天上
play-->notifyAll
play-->wait
觀眾觀看:天上
watch-->notifyAll
watch-->wait
演員表演:人間
play-->notifyAll
play-->wait
觀眾觀看:人間
watch-->notifyAll
watch-->wait
演員表演:天上
play-->notifyAll
play-->wait
觀眾觀看:天上
watch-->notifyAll
watch-->wait
演員表演:人間
play-->notifyAll
play-->wait
觀眾觀看:人間
watch-->notifyAll
watch-->wait
演員表演:天上
play-->notifyAll
play-->wait
觀眾觀看:天上
watch-->notifyAll
watch-->wait
演員表演:人間
play-->notifyAll
play-->wait
觀眾觀看:人間
watch-->notifyAll
watch-->wait
演員表演:天上
play-->notifyAll
play-->wait
觀眾觀看:天上
watch-->notifyAll
watch-->wait
演員表演:人間
play-->notifyAll
play-->wait
觀眾觀看:人間
watch-->notifyAll
watch-->wait
演員表演:天上
play-->notifyAll
play-->wait
觀眾觀看:天上
watch-->notifyAll
watch-->wait
演員表演:人間
play-->notifyAll
觀眾觀看:人間
watch-->notifyAll

執行緒池

  1. 背景:經常建立和銷燬、使用量特別大的資源,比如併發情況下的執行緒,對效能影響很大。
  2. 思路:提前建立好多個執行緒,放入執行緒池中,使用時直接獲取,使用完放回池中。可以避免頻繁的建立銷燬,實現重複利用。
  3. 好處:
    1. 提供響應速度(減少了建立新執行緒的時間)。
    2. 降低資源消耗(重複利用執行緒池中的執行緒,不需要每次都建立)。
    3. 便於執行緒管理
      1. corePoolSize:核心池的大小
      2. maximumPoolSize:最大執行緒數
      3. keepAliveTime:執行緒沒有任務時最多保持多長時間後會終止
  4. JDK5.0提供了執行緒池相關API:ExecutorService和Executors。
  5. ExecutorService:真正的執行緒池介面。常見子類ThreadPoolExecutor.
    1. void execute(Runnable command):執行任務/命令,沒有返回值,一般用來執行Runnable.
    2. Future submit(Callable task):執行任務,有返回值,一般用來執行Callable。
    3. void shutdown:關閉連線池。
  6. Executors:工具類、執行緒池的工廠類,用於建立並返回不同型別的執行緒池。
package com.qing.sync;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 測試執行緒池
 */
public class TestPool {
    public static void main(String[] args) {
        //1.建立服務,建立執行緒池
        //引數:執行緒池大小
        ExecutorService service = Executors.newFixedThreadPool(5);

        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());

        //2.關閉連線
        service.shutdown();
    }
}

class MyThread implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName());
    }
}
pool-1-thread-1
pool-1-thread-2
pool-1-thread-4
pool-1-thread-3
pool-1-thread-3
pool-1-thread-3
pool-1-thread-3
pool-1-thread-5

總結

package com.qing.sync;

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

/**
 * 回顧總結執行緒的建立
 */
public class ThreadNew {
    public static void main(String[] args) {
        new MyThread1().start();

        new Thread(new MyThread2()).start();

        FutureTask<Integer> futureTask = new FutureTask<>(new MyThread3());
        new Thread(futureTask).start();
        try {
            Integer integer = futureTask.get();
            System.out.println(integer);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

//1.繼承Thread類
class MyThread1 extends Thread {
    @Override
    public void run() {
        System.out.println("MyThread1");
    }
}
//2.實現Runnable介面
class MyThread2 implements Runnable {
    @Override
    public void run() {
        System.out.println("MyThread2");
    }
}
//3.實現Callable介面
class MyThread3 implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        System.out.println("MyThread3");
        return 10;
    }
}
MyThread1
MyThread2
MyThread3
10