java多线程,多线程加锁以及Condition类的使用

Posted 2020-11-22 miaoliangjun

看了网上非常多的运行代码,很多都是重复的再说一件事,可能对于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,
",x);
    }
    static void printBuzz(int x){
        System.out.printf("%d:Buzz,
",x);
    }
    static void printFizzBuzz(int x){
        System.out.printf("%d:FizzBuzz,
",x);
    }
    static void printaccpt(int x){
        System.out.printf("%d,
",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);
    }
}

View Code