執行緒協作

執行緒通訊

應用場景：生產者和消費者問題，生產者生產商品，消費者購買商品，沒有商品時等待生產商生產放入商品

分析

執行緒同步問題。生產者和消費者共享同一資源，相互依賴，互為條件：

• 生產者沒有生產產品前，需要通知消費者等待，生產後要通知消費者消費
• 消費者在消費後要通知生產者結束消費，生產新的產品以供消費
• synchronized不足使用：
  • synchronized可以同步
  • 但無法通訊

方法

Java提供了一些Object類的方法解決執行緒間的通訊問題，僅能在同步方法和同步塊中使用，否則會丟擲異常

• wait()：執行緒會一直等待，直到其他執行緒通知，會釋放鎖
• wait(long timeout)
  ：指定等待的毫秒數
• notify()：喚醒一個處於等待狀態的線
• notifyAll(): 喚醒同一個物件上所有地道用wait()方法的執行緒，優先級別高的執行緒優先排程

管程法：利用緩衝區

涉及wait()時使用while

//管程法
//生產者，消費者，產品，緩衝區
public class TestPC {
    public static void main(String[] args) {
        SynContainer container = new SynContainer();

        //測試，多名生產者和多名消費者
        new Producer(container).start();
        new Producer(container).start();
        new Producer(container).start();
        new Consumer(container).start();
        new Consumer(container).start();
        new Consumer(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 = 0; i < 100; i++) {
            container.push(new Food(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 = 0; i < 100; i++) {
            System.out.println("消費了：" + container.pop().id);
        }
    }
}

//產品
class Food{
    int id;

    public Food(int id) {
        this.id = id;
    }
}

//緩衝區
class SynContainer{
    //容器大小
    Food[] foods = new Food[10];
    //容器計數器
    int count = 0;

    //生產者放入產品
    public synchronized void push(Food food){
        //容器滿了，等待消費者消費
        while (count == foods.length){
            //通知消費者消費，生產等待
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //沒滿，生產產品
        foods[count] = food;
        count++;

        //通知消費者消費
        this.notifyAll();
    }

    //消費者消費產品
    public synchronized Food pop(){
        //判斷是否有產品，多名消費者都等待
        while (count <= 0){
            //等待產品生產，消費者等待
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //可以消費
        count--;//出棧，因為生產後索引指向後一個空位
        Food food = foods[count];

        //消耗了，通知生產者生產
        this.notifyAll();
        return food;
    }
}
 

訊號燈法：標誌位

//訊號燈法，標誌位解決
public class TestPC2 {
    public static void main(String[] args) {
        Produce produce = new Produce();
        new Producer2(produce).start();
        new Producer2(produce).start();
        new Consumer2(produce).start();
        new Consumer2(produce).start();
    }
}
//生產者
class Producer2 extends Thread{
    Produce id;

    public Producer2(Produce id) {
        this.id = id;
    }

    //生產
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            this.id.product(i);
        }
    }
}

//消費者
class Consumer2 extends Thread{
    Produce id;

    public Consumer2(Produce id) {
        this.id = id;
    }

    //消費
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            this.id.consume();
        }
    }
}

class Produce{
    //生產者生產，消費者等待 T
    //消費者消費，生產者等待 F
    int id;
    boolean flag = true;

    //生產
    public synchronized void product(int id){
        if (!flag){
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("生產了" + id);
        //通知消費者消費
        this.notifyAll();
        this.id = id;
        this.flag = !this.flag;
    }

    //消費
    public synchronized void consume(){
        if (flag){
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("消費了" + id);

        //通知生產
        this.notifyAll();
        this.flag = !this.flag;
    }
}
 

執行緒池

• 背景：經常建立和銷燬、使用量特別 大的資源，如併發情況下的執行緒，對效能影響很大
• 思路：提前建立好多個執行緒，放入執行緒池中，使用時之間獲取，使用完放回池中。避免頻發建立銷燬，實現重複利用
• 優點：
  • 提高響應速度（減少了建立新執行緒的時間）
  • 降低資源消耗（重複利用執行緒池中執行緒，不需要每次重新建立）
  • 便於執行緒管理：
    • corePoolSize：核心池的大小
    • maximumPoolSize：最大執行緒數
    • keepAliveTime：執行緒沒用任務時最多保留多久中止

執行緒池的使用

• JDK5.0，ExecutorService，Executors
• ExecutorService：執行緒池介面，常用子類ThreadPoolExecutor
  • void execute(Runnable command)：執行任務/命令，沒返回值，用於執行Runnable
  • <T>Future<T> submit(Callable<T> task)：執行任務，有返回值，用於執行Callable
  • void shutdown()：用於關閉連線池
• Executors：工具類、執行緒池的工廠類，用於建立並返回不同型別的執行緒池

public class TestPool {
    public static void main(String[] args) {
        //建立服務，建立執行緒池
        //newFixedThreadPool，引數為執行緒大小
        ExecutorService service = Executors.newFixedThreadPool(10);

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

        //關閉
        service.shutdown();
    }
}
class MyThread implements Runnable{
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName());
    }
}