线程之间的通信

Posted 2022-03-26 zhuchangwu

一. 等待通知机制的实现

方法名	作用
wait()	执行当前代码的线程等待
wait(long timeout)	timeout时间内若没有其他线程唤醒,也会醒过来
wait(long timeout, int nanos)	超出timeout和额外的时间nanos,没有被其他线程唤醒,也会醒过来

方法名	作用
notify()	随机唤醒一条在等待队列中想去访问同一共享变量的线程
void notifyAll()	唤醒在此对象监视器上等待的所有线程

wait()可以使当前线程停下来,等待某个条件发生变化,并且这个条件超出了当前方法的控制范围,可以实现和自旋一样的效果,但是呢,自旋确实一种CPU的不良的使用行为

实例代码:

public class demo1 
    private static List list = new ArrayList();

    public   void  add()
        list.add("string");
    

    public   int size()
        return list.size();
    

public class demo11 

    public static void main(String[] args) 
        demo1 demo1 = new demo1();

        Object o = new Object();

        new Thread(()->
            synchronized (o)
                if(demo1.size()!=5)
                    try 
                        System.out.println("开始等待..."+ System.currentTimeMillis());
                        o.wait();
                        System.out.println("等待结束.."+System.currentTimeMillis());
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
            
        ).start();

        new Thread(()->
            synchronized (o)
                for(int i=0;i<10;i++)
                    demo1.add();
                    if (demo1.size()==5)
                        System.out.println("发出notofy通知...");
                        o.notify();
                    
                    System.out.println("已经添加了"+i+"个元素");

                    try 
                        Thread.sleep(300);
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
            
        ).start();

运行结果:
开始等待...1549259002081
已经添加了0个元素
已经添加了1个元素
已经添加了2个元素
已经添加了3个元素
发出notofy通知...
已经添加了4个元素
已经添加了5个元素
已经添加了6个元素
已经添加了7个元素
已经添加了8个元素
已经添加了9个元素
等待结束..1549259005086

wait() & notify() & notifyAll()

wait()总结:

• wait()方法是Object的方法,作用是让执行当前代码的线程进行等待,(置入到预执行队列),并且会记住当前线程执行到了哪一行代码,当现场被唤醒后,继续从记住的那行代码往下执行
• wait()方法使用的前提,线程必须获取到对象级别的锁,这也就意味着,wait()必须在synchronized同步方法,或者同步代码块中才能执行
  • 若没有获取到对象锁,抛出异常IllegalMonitorStateExeception
• 当调用wait()方法后,会立刻释放当前的对象锁

notify()总结:

• notify()同样是Object的方法,调用此方法的效果是:随机的在唤醒一个等待队列中等待访问同一个共享资源的一个线程
  • notifyAll()唤醒所有,(此时,优先级更高的那个线程,有更大几率先执行,但是也一8I吗确定)
• notify()使用的前提同样也是线程首先获取到对象级别的锁
• 调用notify()后,不会立即释放锁,而是继续执行notify()所在的方法,直到此同步方法执行结束后,才会释放对象锁,这也就意味着,notify()之后,wait()状态的线程不会立即被唤醒.

notifyAll()的调用,必须提前获取到锁,而wait()一经调用,立即释放锁,两者不会冲突

二 解决过早通知问题

• 试想,如果通知过早,那么就会打乱正常的逻辑,wait()也就没必要执行了,因为它永远都醒不了

public class demo2 
    String lock = new String("");
    boolean tag = false;

  private Runnable runnableA = new Runnable()

      @Override
      public void run() 
            synchronized (lock)
                try 
              while(tag==false) 
                  System.out.println("runnableA bagin wait...");
                  lock.wait();
                  System.out.println("RunnableA wait end...");
              
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            

      
  ;

private Runnable runnableB = new Runnable()

    @Override
    public void run() 
        synchronized (lock)
                System.out.println("runnableB bagin notify...");
                lock.notify();
                tag=true;
                System.out.println("RunnableB notify end...");
            System.out.println("不满足条件不唤醒...");
        
    
;


public static void main(String[] args) 
    demo2 demo2 = new demo2();
    new Thread(demo2.runnableB).start();
    new Thread(demo2.runnableA).start();

执行结果:

runnableB bagin notify...  
RunnableB notify end...  
不满足条件不唤醒... 

添加了一个判断的条件,实现,若现进行了唤醒,那么不执行wait()

更换两个线程的启动顺序
运行结果:

E:\JavaJDK\bin\java.exe 
runnableA bagin wait...
runnableB bagin notify...
RunnableB notify end...
不满足条件不唤醒...
RunnableA wait end...

三. 等待wait()的条件发生变化与解决

运行下面的代码:

public class demo33 

    private String lock;
    public  demo33(String lock)
        this.lock =lock;
    
    private List list = new ArrayList();
    public void add()
        synchronized (lock)
            list.add("hello");

            System.out.println(Thread.currentThread().getName()+"add hello 然后唤醒所有wait()线程");
            //唤醒所有
            lock.notifyAll();
        
    

    public void subtract()
        synchronized (lock)
           if(list.size()==0)
               try 
                   System.out.println(Thread.currentThread().getName()+"开始等待");
                   lock.wait();
                   System.out.println(Thread.currentThread().getName() +"等待结束");
                catch (InterruptedException e) 
                   e.printStackTrace();
               
           
           list.remove(0);

        
    


 public static void main(String[] args) 

        demo33 demo33 = new demo33("123");

        // 第一条等待的线程
        new Thread(()->
            demo33.subtract();
        ).start();
        //第二条等待的线程
       new Thread(()->
            demo33.subtract();
        ).start();

        //唤醒所有
        new Thread(()->
            demo33.add();
        ).start();

    

运行结果:

Thread-0开始等待
Thread-1开始等待
Thread-2add hello 然后唤醒所有wait()线程
Thread-1等待结束
Thread-0等待结束
Exception in thread "Thread-0" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
    at java.util.ArrayList.rangeCheck(ArrayList.java:657)

• 抛出了异常,原始就是因为wait的条件发生了改变,前两条线程因为if(list.size()==0)而wait(),紧接着,add()一个元素后,notifyAll()唤醒了所有等待中的线程,于是,那两条等待中的线程在wait()处,继续往下执行remove(0),我们知道,仅仅是添加了一个元素,第二次remove(0)的时候,是非法的,而在我们的判断if(list.size()==0)..中判断反应不过来于是抛出了异常
• 解决方案很简单,既然来不及判断,我们就用while()替换if()这样一来,while()会比if()多执行一次,发现条件满足,继续等待

四. 生产者消费者模式

1 解决多生产多消费: 操作值 -- 假死现象

/*
* 多生产,多消费的假死现象
* */
public class demo4 
private List list = new ArrayList();

public void p(Object o)
    synchronized (o)
        while(!(list.size()==0))  //对于生产者,list不为空,等待
            System.out.println("生产者"+Thread.currentThread().getName()+"等待了...");
            try 
                o.wait();
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
        // 为空,生产
        System.out.println("生产者"+Thread.currentThread().getName()+"生产了...");
        list.add("123");
        o.notify();
    


public void c(Object o)
    synchronized (o)
        while(list.size()==0)  //对于消费者,size==0 等待
            System.out.println("消费者"+Thread.currentThread().getName()+"等待了");
            try 
                o.wait();
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
        System.out.println("消费者"+Thread.currentThread().getName()+"消费了");
        list.remove(0);
        o.notify();
    



public static void main(String[] args) 
    demo4 demo4 = new demo4();
    Object o = new Object();
    ExecutorService executorService = Executors.newCachedThreadPool();
   for (int i=0;i<5;i++)

        executorService.execute(new Runnable() 
            @Override
            public void run() 
                while (true)
                demo4.p(o);
            
        );
        executorService.execute(new Runnable() 
            @Override
            public void run() 
                while (true)
                demo4.c(o);
            
        );
       
  
    Thread[] threads = new Thread[2];
    try 
        Thread.sleep(5000);
        System.out.println("主函数醒了");
     catch (InterruptedException e) 
        e.printStackTrace();
    

运行上面的代码结果

.
.
.
生产者pool-1-thread-7等待了...
消费者pool-1-thread-10消费了
消费者pool-1-thread-10等待了
消费者pool-1-thread-8等待了
消费者pool-1-thread-6等待了
主函数醒了

卡顿在最后"主函数醒了"不再往下运行,产生了假死的现象实际上还有线程依然存活,只不过是它一直等待,没人唤醒它,notify()会随机唤醒一条线程,这也就意味着,可能存在生产者唤醒的是生产者,消费者唤醒了消费者,导致双方全部等待而造成假死

• 解决方法,把notify()换成notifyAll()(它不但会唤醒同己,也会唤醒异己)

2 . 一生产与多消费--操作栈,解决wait条件改变与假死问题

• 解决wait条件改变---使用while替换if进行判断
• 假死的原因依然是唤醒了同类---notifyAll()替换notify()

五 . 方法join

1. 简单使用

很多情况下,主线程中启动子线程,然后两条线程并行运行,主线程往往早于子线程之前结束,那么,假如说主线程想等子线程执行完毕后,拿到子线程的结果后再结束,那么最直接的方法就是 使用join()

public class myjoin extends Thread
    public void run()
        System.out.println("我是子线程,我要睡两秒...");
        try 
            Thread.sleep(2000);
         catch (InterruptedException e) 
            e.printStackTrace();
        
    

    public static void main(String[] args) 

        try 
            myjoin myjoin = new myjoin();
            myjoin.start();
            myjoin.join();
         catch (InterruptedException e) 
            e.printStackTrace();
        
        System.out.println(Thread.currentThread().getName()+"我成功的等待子线程执行完后,执行");
    

运行结果

我是子线程,我要睡两秒...
main我成功的等待子线程执行完后,执行

• join方法的作用就是,执行当前线程的run()中的任务,直到任务结束,线程对象销毁后,才执行后面的代码

2. join方法与异常

• 当join()所在的线程碰到了interrupted()会摩擦出怎样的火花呢?

观看如下代码

public class ThreadA extends Thread  
public void run()
    try 
        System.out.println("ThreadA 执行了,紧接着睡五秒...");
        Thread.sleep(5000);
        System.out.println("五秒了, ThreadA 醒过来...");
     catch (InterruptedException e) 
        e.printStackTrace();
    






public class ThreadB extends Thread

public void run()
        try
            System.out.println("ThreadB 启动了...");
            ThreadA threadA = new ThreadA();
            threadA.start();
            threadA.join();
            System.out.println("ThreadA join 之后的代码...");
        catch (Exception e)
            System.out.println("ThreadB catch块打印了...");
            e.printStackTrace();
        





public class ThreadC extends Thread 

Thread thread;
public ThreadC(Thread b)
    this.thread=b;


public void run()
 thread.interrupt();


public static void main(String[] args) 

    ThreadB threadB = new ThreadB();
    threadB.start();

    ThreadC threadC = new ThreadC(threadB);
    threadC.start();
    System.out.println("主线程结束...");

运行结果:

ThreadB 启动了...
主线程结束...
java.lang.InterruptedException
ThreadB 启动了...
ThreadB catch块打印了...
ThreadA 执行了,紧接着睡五秒...
    at java.lang.Object.wait(Native Method)
    at java.lang.Thread.join(Thread.java:1252)
    at java.lang.Thread.join(Thread.java:1326)
    at com.atGongDa.MultiThreading.线程之间的通信.ThreadB.run(ThreadB.java:10)
五秒了, ThreadA 醒过来...

• 通过出现了中断异常,原因是ThreadA还在运行,并且没出现异常
• System.out.println("ThreadA join 之后的代码..."); 并未输出, ThreadB确实被中断了
• ThreadA正常执行结束

当时对join()阻塞的是哪条线程的代码还是有点模糊,现在分析结果,可以看到,join阻塞的是join()方法 所在的那条线程,根据上面的例子,join只能阻塞ThreadB,却不能阻塞ThreadC 和 主线程

3. 意外: join()后面的代码提前执行现象与解释

运行下面的代码:

public class ThreadQ extends Thread
    private Thread W;

    public ThreadQ(Thread thread) 
        this.W = thread;
    
    public void run()
        synchronized (W)
            System.out.println("ThreadQ准备开始睡三秒..."+System.currentTimeMillis());
            try 
                Thread.sleep(3000);
             catch (InterruptedException e) 
                e.printStackTrace();
            
            System.out.println("ThreadQ...睡醒了..."+System.currentTimeMillis());
        
    




public class ThreadW extends Thread 

synchronized public void run() 
   try 
       System.out.println("ThreadW 启动了...要睡三秒"+System.currentTimeMillis());
       Thread.sleep(3000);
       System.out.println("ThreadW睡醒了..."+System.currentTimeMillis());
        catch (InterruptedException e) 
       e.printStackTrace();
   





public class ThreadE    

public static void main(String[] args) 
    ThreadW w = new ThreadW();
    ThreadQ threadQ = new ThreadQ(w);

    w.start();
    threadQ.start();

    try 
        threadQ.join(2000);
     catch (InterruptedException e) 
        e.printStackTrace();
    

    System.out.println("main end...");

运行结果: 可以看到,main end提前打印出来了

ThreadW 启动了...要睡三秒1549362922247
main end...
ThreadW睡醒了...1549362925248
ThreadQ准备开始睡三秒...1549362925248
ThreadQ...睡醒了...1549362928248

当我们把join(2000)注释掉后结果如下

main end...1549363120715
ThreadQ准备开始睡三秒...1549363120715
ThreadQ...睡醒了...1549363123715
ThreadW 启动了...要睡三秒1549363123715
ThreadW睡醒了...1549363126716

分析结果,不难看出,在线程的启动一条新的线程比它运行自己的代码要快的多.因此,大多数情况下,都是join()方法先执行,拿到对象锁,然后马上释放掉...,然后Q抢到ThreadW对象锁,睡上三秒且不释放,ThreadW因为没有锁,故执行不了自己加上了synchronize的run()方法,本来join()可以阻塞Zhu后面的代码,可是join(2000),发现自己已经过期了,因此ThreadW和ThreadMain就会异步执行

六. join() & sleep()的区别

• join()方法底层是wait()实现的,这也就意味着,当我们调用join()方法时,它会做两件事
  • 阻塞join()方法所在的线程1,执行调用join()方法的线程2的run任务

  • 释放掉调用join方法的线程2对象的 对象锁

    这也就意味着,其他的线程可以访问调用join方法的线程2的同步方法...

• sleep()不会释放,运行当前代码的线程对象的 对象锁,也就是说,其他线程是不能访问此线程的其他同步方法的

七 ThreadLocal

变量值的共享可以使用public static 修饰,所有的线程都使用同一个public static 的变量,如果想实现每一个线程都有自己的共享变量呢? ThreadLocal ,可以把它当成一个专属于当前线程对象的盒子,它保证了线程之前的隔离性

• 当我们在run方法中,直接new ThreadLocal对象的时候,get()出来的默认值为null

重写 initialValue值设置默认值

@Override
protected Object initialValue() 
    return "123";

八 InheritableThreadLocal

• 共用一套工具InheritableThreadLocal,实现了让子线程从父线程中获取值,

重写childValue方法,实现继承值的修改,

@Override
protected Object childValue() 
    return "XXX";

注意:
当子线程从父线程中获取值的同时,父线程把值修改了,子线程获取到的值为旧值

---

> 在java jdk1.5开始,java平台提供了更高级的并发工具,他可以完成以前必须在wait()和notify()上手写代码来完成的各项工作,这在一定程度上让我们几乎没有任何理由再去使用wait和notify,这也是<>提及的第69条,并发工具1.Executor Framework 2,Concurrent Collection 3同步器 Synchronizer ,优先于wait notify

参考书籍<>高洪岩著 <>