java多線程,多線程加鎖以及Condition類的使用

看了網上非常多的運行代碼,很多都是重復的再說一件事,可能對於java老鳥來說,理解java的多線程是非常容易的事情,但是對於我這樣的菜鳥來說,這個實在有點難,可能是我太菜了,網上重復的陳述對於我理解這個問題一點幫助都沒有.所以這里我寫下我對於這個問題的理解,目的是為了防止我忘記.

還是從代碼實例開始講起:

import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        MyBlockingQueue<Integer> queue = new MyBlockingQueue<>(1);
        for (int i = 0; i < 10; i++) {
            int data = i;
            new Thread(() -> {
                try {
                    queue.enqueue(data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        System.out.println("1111111");
        for(int i=0;i<10;i++){
            new Thread(() -> {
                try {
                    queue.dequeue();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }).start();
        }
    }
    public static class MyBlockingQueue<E> {
        int size;//阻塞隊列最大容量
        ReentrantLock lock = new ReentrantLock(true);
        LinkedList<E> list=new LinkedList<>();//隊列底層實現
        Condition notFull = lock.newCondition();//隊列滿時的等待條件
        Condition notEmpty = lock.newCondition();//隊列空時的等待條件
        public MyBlockingQueue(int size) {
            this.size = size;
        }
        public void enqueue(E e) throws InterruptedException {
            lock.lock();
            try {
                while(list.size() ==size)//隊列已滿,在notFull條件上等待
                    notFull.await();

                list.add(e);//入隊:加入鏈表末尾
                System.out.println("入隊：" +e);
                notEmpty.signal(); //通知在notEmpty條件上等待的線程
            } finally {
                lock.unlock();
            }
        }
        public E dequeue() throws InterruptedException {
            E e;
            lock.lock();
            try {
                while(list.size() == 0)
                    notEmpty.await();
                e = list.removeFirst();//出隊:移除鏈表首元素
                System.out.println("出隊："+e);
                notFull.signal();//通知在notFull條件上等待的線程
                return e;
            } finally {
                lock.unlock();
            }
        }
    }
}

主函數啟動了20個線程,前10個是入隊的后10個是出隊的,我們可以看啊可能輸出結果,

new Thread(() -> {
    try {
        queue.enqueue(data);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}).start(); 注意到線程實現,這個是lambda表達式實現Runable接口.

入隊：0
1111111
出隊：0
入隊：2
出隊：2
入隊：1
出隊：1
入隊：3
出隊：3
入隊：4
出隊：4
入隊：5
出隊：5
入隊：6
出隊：6
入隊：7
出隊：7
入隊：8
出隊：8
入隊：9
出隊：9

可以看到1111111在第一個出隊之前,隊列容量為1,也就是說頭10個入隊進程只有第一個成功了,其他均被阻塞.

並且出隊入隊順序是按照循環順序的,說明鎖是按照請求順序來獲取的,先到先得,這個說的就是公平鎖的意思,其實ReentrantLock既可以是公平鎖也可以是非公平鎖,其初始化的時候,往構造函數里面傳入true則為公平鎖,false則為非公平鎖.

至於什么是可重入鎖,可以看看這篇博客https://blog.csdn.net/qq_38737992/article/details/90613821,這個是可重入鎖的實例.

其中有一行代碼很奇怪,lock.lock();對於我這樣的萌新很好奇這個一行代碼到底發生了什么,網上很多都是說獲得鎖,但是"獲得"這個實在難以太不具體,所以我自己想象了一下,感覺大致上就是這樣的一張圖:

 

結合我粗淺的經驗猜測:jvm只有一個就緒隊列,就緒隊列里面的線程按照隊列順序使用cpu資源,若不加鎖,那么所有線程都可以按序取得資源,但是由於面向對象了,所以不好直接控制就緒隊列里面線程的入隊順序,這個時候就需要加鎖來控制線程的運行順序來保證處理邏輯正確.(其實也不不是那么嚴格的隊列,就緒狀態的線程如果是非公平鎖一般會隨機先后的運行,說是隊列而已,其實就是表達就緒狀態)

結合代碼的lock()方法,如果有線程進入cpu並且調用lock(),如果該鎖沒有被其他線程獲取過,那么這個線程可以使用cpu時間,如果該鎖已經被其他線程獲取了,那么該線程會給阻塞,進入阻塞隊列, 這樣來說的話,其實"獲取"這個詞也沒什么難以理解的,其實就是一個標記而已,然后lock()方法其實就只是判斷當前線程是使用cpu時間,還是進入阻塞隊列而已..

在看看unlock()方法,由上面的圖幫助,其實這個也很好理解,其實就是把阻塞隊列的隊首的線程出隊,然后進入就緒隊列而已.

可以猜測,如果運行過程中有多個鎖實例,那么就會有多少個可能阻塞的線程,那么除了使用用多個鎖,其實還有別的方法來增加阻塞線程,就是使用Condition類,需要指出的是condition類的await()方法,會阻塞當前線程,然后自動解除當前線程獲取的鎖(這點尤其重要),切換線程,如果其他線程中有喚醒,那么這個在被喚醒后線程會從await()的位置繼續往下運行,所以一般要配合while循環使用,如果某線程被喚醒,那么它對於它之前獲取的鎖,也將重新獲取,如果此時該鎖已經被另外一個線程獲取,且還沒有解鎖,此時的喚醒就會出錯,會出現莫名其妙的錯誤,所以需要設置一個volatile變量來檢測線程的運行狀態,所以await()方法前后都要檢測.

這里提出一道題,來自leetcode,要求使用condition類來寫,

題意:

編寫一個可以從 1 到 n 輸出代表這個數字的字符串的程序，但是：

如果這個數字可以被 3 整除，輸出 "fizz"。
如果這個數字可以被 5 整除，輸出 "buzz"。
如果這個數字可以同時被 3 和 5 整除，輸出 "fizzbuzz"。
例如，當 n = 15，輸出： 1, 2, fizz, 4, buzz, fizz, 7, 8, fizz, buzz, 11, fizz, 13, 14, fizzbuzz。

來源：力扣（LeetCode）
鏈接：https://leetcode-cn.com/problems/fizz-buzz-multithreaded

 解法:

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

public class Main {
    static void printFizz(int x){
        System.out.printf("%d:Fizz,\n",x);
    }
    static void printBuzz(int x){
        System.out.printf("%d:Buzz,\n",x);
    }
    static void printFizzBuzz(int x){
        System.out.printf("%d:FizzBuzz,\n",x);
    }
    static void printaccpt(int x){
        System.out.printf("%d,\n",x);
    }
    static volatile int now=1;
    static ReentrantLock lock=new ReentrantLock();
    static Condition k1=lock.newCondition();
    public static void test(int n) {

        new Thread(()->{
            while(now<=n){
                lock.lock();
                try{
                    while(now%5==0||now%3!=0){
                        if(now>n) throw new InterruptedException();
                        k1.await();
                        if(now>n) throw new InterruptedException();
                    }
                    printFizz(now);
                    now++;
                    k1.signalAll();
                } catch (InterruptedException e) {
                    break;
                    //e.printStackTrace();
                } finally{
                    lock.unlock();
                }
            }
            System.out.println("Thread 1 is over");
        }).start();

        new Thread(()->{
            while(now<=n){
                lock.lock();
                try{

                    while(now%5!=0||now%3==0) {
                        if(now>n) throw new InterruptedException();
                        k1.await();
                        if(now>n) throw new InterruptedException();
                    }
                    printBuzz(now);
                    now++;
                    k1.signalAll();
                } catch (InterruptedException e) {
                    break;
                   // e.printStackTrace();
                } finally{
                    lock.unlock();
                }
            }
            System.out.println("Thread 2 is over");
        }).start();


        new Thread(()->{
            while(now<=n){
                lock.lock();
                try{
                    while(now%5!=0||now%3!=0) {
                        if(now>n) throw new InterruptedException();
                        k1.await();
                        if(now>n) throw new InterruptedException();
                    }
                    printFizzBuzz(now);
                    now++;
                    k1.signalAll();
                } catch (InterruptedException e) {
                    break;
                    //Thread.interrupted();
                    //e.printStackTrace();
                } finally{
                    lock.unlock();
                }
            }
            System.out.println("Thread 3 is over");
        }).start();


        new Thread(()->{
            while(now<=n){
                lock.lock();
                try{
                    while(now%5==0||now%3==0) {
                        if(now>n) throw new InterruptedException();
                        k1.await();
                        if(now>n) throw new InterruptedException();
                    }
                    printaccpt(now);
                    now++;
                    k1.signalAll();
                }catch (InterruptedException e){
                    break;
                    //Thread.interrupted();
                    //e.printStackTrace();
                }
                finally{
                    lock.unlock();
                }
            }
            System.out.println("Thread 4 is over");
        }).start();
    }

    public static void main(String[] args) throws InterruptedException {
        test(30);
    }
}