线程安全5 - Lock & Condition实现同步通信

作者: 小超_8b2f | 来源:发表于2019-09-29 11:06 被阅读0次

Lock比传统线程模型中的synchronized方式更加面向对象，与生活中的锁类似，锁本身也应该是一个对象。两个线程执行的代码片段要实现同步互斥的效果，他们必须用同一个Lock对象。锁是上在代表要操作资源的类的内部方法中，而不是线程代码中！
读写锁：分为读锁和写锁，多个读锁不互斥，读锁和写锁互斥，写锁与写锁互斥。这是JVM自己控制的，你只要上好相应的锁即可。如果你的代码只读数据，可以多人同时读，但不能同时写，那就上读锁；如果你的代码修改数据，只能一个人在写，且不能同时读取，那就上写锁。总之，读的时候上读锁，写的时候上写锁！
Condition的功能类似在传统线程技术中的Object.wait和Object.notify的功能。在等待Condition时，允许发生“虚假唤醒”，这通常作为基础平台语义的让步。对于大多数应用程序，这带来的实际影响很小，因为Condition应该总是在一个循环中被等待，并测试整备等待的状态声明。某个实现可以随意移除可能的虚假唤醒，但建议应用程序程序员总是假定这些虚假唤醒可能发生，因此总是在一个循环中等待。
一个锁内部可以有多个Condition，即有多路等待和通知，可以看jdk1.5提供的Lock和Condition实现的可阻塞队列的应用案例，从中除了要体味算法，还要体味面向对象的封装。在传统的线程机制中一个监视器对象上只能有一路等待通知，要想实现多路等待通知，必须嵌套使用多个同步监视器对象。（如果只用一个Condition，两个放的都在等，一旦一个放的进去了，那么它通知可能会导致另一个放接着往下走）

1. wait notify 实例：交替执行版本（✔️）

public class MainSubThread {
    public static void main(String[] args) {
        final Business business = new Business();
        new Thread(new Runnable() {
            @Override
            public void run() {
                //子线程循环调用50次某逻辑
                for(int i = 0; i < 50; i++) {
                    business.sub(i);
                }
            }
        }).start();
        
        //主线程循环调用50次某逻辑
        for(int i = 0; i < 50; i++) {
            business.main(i);
        }
    }
}

class Business {
    private boolean bShouldSub = true;
    public synchronized void sub(int i) {
        if(!bShouldSub) {  //while 更健壮，根据API,可以防止伪唤醒
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        for (int j = 0; j < 10; j++)
            System.out.println("sub Thread run sequece of " + j + " loop of " + i);
        
        bShouldSub = false;
        this.notify();
    }
    
    public synchronized void main(int i) {
        if(bShouldSub) {//while 更健壮，根据API,可以防止伪唤醒
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        for (int j = 0; j < 10; j++)
            System.out.println("main Thread run sequece of " + j + " loop of " + i);
        
        bShouldSub = true;
        this.notify();
    }
}

2. Lock代替synchronized，报错(❌)

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

public class MainSubThread {
    public static void main(String[] args) {
         final Business b = new Business();
         
         new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++)
                    b.sub(i);
            }
        }).start();
         
         new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++)
                    b.main(i);
            }
        }).start();
    }
}

class Business {
    private boolean bShouldSub = true;
    Lock lock = new ReentrantLock();
    public void sub(int i) {
        lock.lock();
        try {
            while(!bShouldSub) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            for(int j = 0; j < 10; j++)
                System.out.println("sub Thread run sequece of " + j + " loop of " + i);
            bShouldSub = false;
            this.notify();
        } finally {
            lock.unlock();
        }
    }
    
    public void main(int i) {
        lock.lock();
        try {
            while(bShouldSub) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            for(int j = 0; j < 10; j++)
                System.out.println("main Thread run sequece of " + j + " loop of " + i);
            bShouldSub = true;
            this.notify();
        } finally {
            lock.unlock();
        }
    }
}

报错原因分析：

wait()和notify()必须在synchronized代码块里写。
synchronized的是哪个对象，调用wait()和notify()的就必须是哪个对象。
eg:在方法声明上写synchronized,调用this.wait() 和this.notify
synchronized(obj) { obj.wait(); obj.notify();}

3. 改正后不再报错，但是不能一替一执行。（去掉wait()、notify() ）

class Business {
    Lock lock = new ReentrantLock();
    public void sub(int i) {
        lock.lock();
        try {
            for(int j = 0; j < 10; j++)
                System.out.println("sub Thread run sequece of " + j + " loop of " + i);
        } finally {
            lock.unlock();
        }
    }
    
    public void main(int i) {
        lock.lock();
        try {
            for(int j = 0; j < 10; j++)
                System.out.println("main Thread run sequece of " + j + " loop of " + i);
        } finally {
            lock.unlock();
        }
    }
}

4. 继续改进：Condition.await() Condition.signal()替代synchronized(对象)的wait、notify

class Business {
    private boolean bShouldSub = true;
    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();
    public void sub(int i) {
        lock.lock();
        try {
            while(!bShouldSub) {
                try {
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            for(int j = 0; j < 10; j++)
                System.out.println("sub Thread run sequece of " + j + " loop of " + i);
            bShouldSub = false;
            condition.signal();
        } finally {
            lock.unlock();
        }
    }
    
    public void main(int i) {
        lock.lock();
        try {
            while(bShouldSub) {
                try {
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            for(int j = 0; j < 10; j++)
                System.out.println("main Thread run sequece of " + j + " loop of " + i);
            bShouldSub = true;
            this.notify();
        } finally {
            condition.signal();
        }
    }
}

5. 既然都是实现同样的功能，有synchronized + wait + notify了，为什么还要有Lock + Condition.await() + Condition.signal() ?

因为能实现wait()、notify()实现不了的功能：

6. JAVA API ：Condition 使用实例:阻塞队列

class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition();  
   final Condition notEmpty = lock.newCondition(); 
//如果

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock();
     try {
       while (count == items.length)
         notFull.await();
       items[putptr] = x;
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal(); //只唤醒取数据的线程
     } finally {
       lock.unlock();
     }
   }

   public Object take() throws InterruptedException {
     lock.lock();
     try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal(); //只唤醒放数据的线程
       return x;
     } finally {
       lock.unlock();
     }
   }
 }

$\color{blue}{为什么用2个Condition？}$

因为线程分2种：放数据线程、取数据线程。
若只有一个Condition来await和signal，那么当取数据时,若队列已空无数据可取时，notify将无差别地唤醒取数据线程和放数据线程中的某一个线程，
可能唤醒的还是取数据线程，已经空了你还取个啥？
同理，当线程已满时，若只有一个Condition，那么将无差别地唤醒所有取数据线程和放数据线程中的某一个线程，那么有可能唤醒的还是放数据线程。已经满了你还放个啥？

总结：

Lock + 【一个Condition + await + signal】 = synchronized + wait + notify = 只能单一通知，没得选
Lock + 【多个Condition + await + signal】= 定制化通知，想通知哪个就通知哪个

7. 多路Condition实例：线程逻辑按某特定顺序执行

老大执行完老二执行，老二执行完老三执行，老三执行完老大执行，如此往复。

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

public class ThreeStep {
    public static void main(String[] args) {
        final StepConditionLogic b = new StepConditionLogic();
         
        new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++)
                    b.step1(i);
            }
        }).start();
         
        new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++)
                    b.step2(i);
            }
        }).start();
         
        new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++)
                    b.step3(i);
            }
        }).start();
    }
}


class StepConditionLogic {
    private int step = 1; //步骤
    Lock lock = new ReentrantLock();
    Condition condition1 = lock.newCondition();
    Condition condition2 = lock.newCondition();
    Condition condition3 = lock.newCondition();
    
    
    
    public void step1(int i) {
        lock.lock();
        try {
            while(step != 1) //尚未走到step1
                condition1.await();//阻塞step1逻辑被相应的线程执行
            
            for(int j = 0; j < 10; j++)
                System.out.println("step1 Thread run sequece of " + j + " loop of " + i);
            step = 2; //step1逻辑执行完成，step进入2
            condition2.signal(); //step1执行完了，通知step2执行
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    public void step2(int i) {
        lock.lock();
        try {
            while(step != 2)//尚未走到step2
                condition2.await(); //阻塞step2逻辑被相应的线程执行
            
            for(int j = 0; j < 10; j++)
                System.out.println("step2 Thread run sequece of " + j + " loop of " + i);
            step = 3; //step2逻辑执行完成，step进入3
            condition3.signal(); //step2执行完了，通知step3执行
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    
    public void step3(int i) {
        lock.lock();
        try {
            while(step != 3) //尚未走到step3
                condition3.await();//阻塞step3逻辑被相应的线程执行
            
            for(int j = 0; j < 10; j++)
                System.out.println("step3 Thread run sequece of " + j + " loop of " + i);
            step = 1; //step3逻辑执行完毕，step进入下一步骤
            condition1.signal(); //step3逻辑执行完成，step进入1
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

线程安全5 - Lock & Condition实现同步通信

1. wait notify 实例：交替执行版本（✔️）

2. Lock代替synchronized，报错(❌)

报错原因分析：

3. 改正后不再报错，但是不能一替一执行。（去掉wait()、notify() ）

4. 继续改进：Condition.await() Condition.signal()替代synchronized(对象)的wait、notify

5. 既然都是实现同样的功能，有synchronized + wait + notify了，为什么还要有Lock + Condition.await() + Condition.signal() ?

6. JAVA API ：Condition 使用实例:阻塞队列

$\color{blue}{为什么用2个Condition？}$

总结：

7. 多路Condition实例：线程逻辑按某特定顺序执行

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

多线程