Java多线程（六）：wait()，notify()和notifyAll()

wait()，notify()和notifyAll()介绍

1.wait()

使当前线程等待，直到另一个线程调用notify(),notifyAll()或者中断，当前线程调用wait()之前必须持有锁，调用wait()之后会释放锁。等到当前线程重新获得锁，恢复执行。

只能在同步方法调用wait()。

2.notify()

唤醒一个正在等待锁的线程，如果有多个线程在等待锁，那么任意唤醒其中一个wait的线程。被唤醒的线程(wait的线程)只有在当前线程(notify的线程)释放锁后才可以执行。

只能在同步方法调用notify()。

3.notifyAll()

唤醒所有正在等待锁的线程。

只能在同步方法调用notifyAll()。

native wait()

openjdk找到src/share/native/java/lang/Object.c

static JNINativeMethod methods[] = {

    {"hashCode",    "()I",                    (void *)&JVM_IHashCode},

    {"wait",        "(J)V",                   (void *)&JVM_MonitorWait},

    {"notify",      "()V",                    (void *)&JVM_MonitorNotify},

    {"notifyAll",   "()V",                    (void *)&JVM_MonitorNotifyAll},

    {"clone",       "()Ljava/lang/Object;",   (void *)&JVM_Clone},

}

我们先看下wait方法

src/share/vm/prims/jvm.cpp

搜索JVM_MonitorWait

JVM_ENTRY(void, JVM_MonitorWait(JNIEnv* env, jobject handle, jlong ms))

  JVMWrapper("JVM_MonitorWait");

  Handle obj(THREAD, JNIHandles::resolve_non_null(handle));

  JavaThreadInObjectWaitState jtiows(thread, ms != 0);

  if (JvmtiExport::should_post_monitor_wait()) {

    //当前线程拥有锁，尚且没有加入到等待队列，所以要推迟wait()

    JvmtiExport::post_monitor_wait((JavaThread *)THREAD, (oop)obj(), ms);

  }

    //进入等待状态

  ObjectSynchronizer::wait(obj, ms, CHECK);

JVM_END

我们看下ObjectSynchronizer::wait方法

src/share/vm/runtime/synchronizer.cpp

// NOTE: must use heavy weight monitor to handle wait()

void ObjectSynchronizer::wait(Handle obj, jlong millis, TRAPS) {

    //偏向锁校验

  if (UseBiasedLocking) {

    //撤销偏向锁

    BiasedLocking::revoke_and_rebias(obj, false, THREAD);

    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");

  }

    //时间校验

  if (millis < 0) {

    TEVENT (wait - throw IAX) ;

    THROW_MSG(vmSymbols::java_lang_IllegalArgumentException(), "timeout value is negative");

  }

   //膨胀为重量级锁

  ObjectMonitor* monitor = ObjectSynchronizer::inflate(THREAD, obj());

  DTRACE_MONITOR_WAIT_PROBE(monitor, obj(), THREAD, millis);

  monitor->wait(millis, true, THREAD);

  dtrace_waited_probe(monitor, obj, THREAD);

}

我们看下monitor->wait方法

src/share/vm/runtime/objectMonitor.cpp

void ObjectMonitor::wait(jlong millis, bool interruptible, TRAPS) {

   Thread * const Self = THREAD ;

   assert(Self->is_Java_thread(), "Must be Java thread!");

   JavaThread *jt = (JavaThread *)THREAD;

   DeferredInitialize () ;

   // Throw IMSX or IEX.

   CHECK_OWNER();

   // check for a pending interrupt 是否有中断信号

   if (interruptible && Thread::is_interrupted(Self, true) && !HAS_PENDING_EXCEPTION) {

     // post monitor waited event.  Note that this is past-tense, we are done waiting.

     if (JvmtiExport::should_post_monitor_waited()) {

        // Note: 'false' parameter is passed here because the

        // wait was not timed out due to thread interrupt.

        JvmtiExport::post_monitor_waited(jt, this, false);

     }

     TEVENT (Wait - Throw IEX) ;

     THROW(vmSymbols::java_lang_InterruptedException());

     return ;

   }

   TEVENT (Wait) ;

   assert (Self->_Stalled == 0, "invariant") ;

   Self->_Stalled = intptr_t(this) ;

   //设置线程的监视锁

   jt->set_current_waiting_monitor(this);

   // create a node to be put into the queue

   // Critically, after we reset() the event but prior to park(), we must check

   // for a pending interrupt.

   //添加一个节点放入到等待队列中

   ObjectWaiter node(Self);

   node.TState = ObjectWaiter::TS_WAIT ;

   Self->_ParkEvent->reset() ;

   OrderAccess::fence();          // ST into Event; membar ; LD interrupted-flag

   //获取锁

   Thread::SpinAcquire (&_WaitSetLock, "WaitSet - add") ;

   //添加节点

   AddWaiter (&node) ;

   //释放锁

   Thread::SpinRelease (&_WaitSetLock) ;

   if ((SyncFlags & 4) == 0) {

      _Responsible = NULL ;

   }

   intptr_t save = _recursions; // record the old recursion count

   //增加等待线程数

   _waiters++;                  // increment the number of waiters

   _recursions = 0;             // set the recursion level to be 1

   exit (Self) ;                    // exit the monitor

   guarantee (_owner != Self, "invariant") ;

   ......

}

native notify()

我们看下notify()

src/share/vm/prims/jvm.cpp

JVM_ENTRY(void, JVM_MonitorNotify(JNIEnv* env, jobject handle))

  JVMWrapper("JVM_MonitorNotify");

  Handle obj(THREAD, JNIHandles::resolve_non_null(handle));

  ObjectSynchronizer::notify(obj, CHECK);

JVM_END

src/share/vm/runtime/synchronizer.cpp

void ObjectSynchronizer::notify(Handle obj, TRAPS) {

    //偏向锁校验

 if (UseBiasedLocking) {

    BiasedLocking::revoke_and_rebias(obj, false, THREAD);

    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");

  }

  markOop mark = obj->mark();

    //检验线程是否有锁

  if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {

    return;

  }

  ObjectSynchronizer::inflate(THREAD, obj())->notify(THREAD);

}

src/share/vm/runtime/objectMonitor.cpp

void ObjectMonitor::notify(TRAPS) {

  CHECK_OWNER();

  if (_WaitSet == NULL) {

     TEVENT (Empty-Notify) ;

     return ;

  }

  DTRACE_MONITOR_PROBE(notify, this, object(), THREAD);

  int Policy = Knob_MoveNotifyee ;

  Thread::SpinAcquire (&_WaitSetLock, "WaitSet - notify") ;

  ObjectWaiter * iterator = DequeueWaiter() ;

  if (iterator != NULL) {

     TEVENT (Notify1 - Transfer) ;

     guarantee (iterator->TState == ObjectWaiter::TS_WAIT, "invariant") ;

     guarantee (iterator->_notified == 0, "invariant") ;

     if (Policy != 4) {

        iterator->TState = ObjectWaiter::TS_ENTER ;

     }

     iterator->_notified = 1 ;

     Thread * Self = THREAD;

     iterator->_notifier_tid = Self->osthread()->thread_id();

     ObjectWaiter * List = _EntryList ;

     if (List != NULL) {

        assert (List->_prev == NULL, "invariant") ;

        assert (List->TState == ObjectWaiter::TS_ENTER, "invariant") ;

        assert (List != iterator, "invariant") ;

     }

......

  }

    //释放锁

  Thread::SpinRelease (&_WaitSetLock) ;

  if (iterator != NULL && ObjectMonitor::_sync_Notifications != NULL) {

     ObjectMonitor::_sync_Notifications->inc() ;

  }

}

native notifyAll()

我们看下notifyAll

和notify()相似

src/share/vm/prims/jvm.cpp

JVM_ENTRY(void, JVM_MonitorNotifyAll(JNIEnv* env, jobject handle))

  JVMWrapper("JVM_MonitorNotifyAll");

  Handle obj(THREAD, JNIHandles::resolve_non_null(handle));

  ObjectSynchronizer::notifyall(obj, CHECK);

JVM_END

我们看下ObjectSynchronizer::notifyall方法

和notify()相似

void ObjectSynchronizer::notifyall(Handle obj, TRAPS) {

    //偏向锁校验

  if (UseBiasedLocking) {

    BiasedLocking::revoke_and_rebias(obj, false, THREAD);

    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");

  }

  markOop mark = obj->mark();

    //检测线程是否有锁

  if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {

    return;

  }

  ObjectSynchronizer::inflate(THREAD, obj())->notifyAll(THREAD);

}

src/share/vm/runtime/objectMonitor.cpp

比notify()多了等待队列

void ObjectMonitor::notifyAll(TRAPS) {

  CHECK_OWNER();

  ObjectWaiter* iterator;

  if (_WaitSet == NULL) {

      TEVENT (Empty-NotifyAll) ;

      return ;

  }

  DTRACE_MONITOR_PROBE(notifyAll, this, object(), THREAD);

  int Policy = Knob_MoveNotifyee ;

  int Tally = 0 ;

  Thread::SpinAcquire (&_WaitSetLock, "WaitSet - notifyall") ;

  for (;;) {

     iterator = DequeueWaiter () ;

     if (iterator == NULL) break ;

     TEVENT (NotifyAll - Transfer1) ;

     ++Tally ;

     // Disposition - what might we do with iterator ?

     // a.  add it directly to the EntryList - either tail or head.

     // b.  push it onto the front of the _cxq.

     // For now we use (a).

     guarantee (iterator->TState == ObjectWaiter::TS_WAIT, "invariant") ;

     guarantee (iterator->_notified == 0, "invariant") ;

     iterator->_notified = 1 ;

     Thread * Self = THREAD;

     iterator->_notifier_tid = Self->osthread()->thread_id();

     if (Policy != 4) {

        iterator->TState = ObjectWaiter::TS_ENTER ;

     }

     ObjectWaiter * List = _EntryList ;

     if (List != NULL) {

        assert (List->_prev == NULL, "invariant") ;

        assert (List->TState == ObjectWaiter::TS_ENTER, "invariant") ;

        assert (List != iterator, "invariant") ;

     }

   ......

     if (Policy < 4) {

       iterator->wait_reenter_begin(this);

     }

  }

   //释放锁

  Thread::SpinRelease (&_WaitSetLock) ;

  if (Tally != 0 && ObjectMonitor::_sync_Notifications != NULL) {

     ObjectMonitor::_sync_Notifications->inc(Tally) ;

  }

}

wait()和notfiy()实例

MyThread30_0类，使用wait()

public class MyThread30_0 extends Thread {

    private Object lock;

    public MyThread30_0(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        try

        {

            synchronized (lock)

            {

                System.out.println("开始------wait time = " + System.currentTimeMillis());

                lock.wait();

                System.out.println("结束------wait time = " + System.currentTimeMillis());

            }

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

    }

}

MyThread30_1类，使用notify()

public class MyThread30_1 extends Thread {

    private Object lock;

    public MyThread30_1(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        synchronized (lock)

        {

            System.out.println("开始------notify time = " + System.currentTimeMillis());

            lock.notify();

            System.out.println("结束------notify time = " + System.currentTimeMillis());

        }

    }

}

main方法,sleep保证mt0线程先执行。

public class MyThread30_main {

    public static void main(String[] args) throws Exception

    {

        Object lock = new Object();

        MyThread30_0 mt0 = new MyThread30_0(lock);

        mt0.start();

        Thread.sleep(3000);

        MyThread30_1 mt1 = new MyThread30_1(lock);

        mt1.start();

    }

}

运行结果如下

开始------wait time = 1563183002681

开始------notify time = 1563183005683

结束------notify time = 1563183005683

结束------wait time = 1563183005683

先执行“开始------wait time = 1563183002681”，lock.wait()会释放锁，mt0线程进入等待状态。

mt1线程获得锁，执行“开始------notify time = 1563183005683”，lock.notify()唤醒wait的线程(mt0)，使mt0退出等待状态，进入可运行状态，mt0线程想要执行必须等待notify的线程(mt1)释放锁，所以执行“结束------notify time = 1563183005683”

最后再恢复执行mt0线程，输出“结束------wait time = 1563183005683”

wait()释放锁

ThreadDomain31类,调用wait方法

public class ThreadDomain31 {

    public void testMethod(Object lock)

    {

        try

        {

            synchronized (lock)

            {

                System.out.println(Thread.currentThread().getName() + " Begin wait()");

                lock.wait();

                System.out.println(Thread.currentThread().getName() + " End wait");

            }

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

    }

}

MyThread31类

public class MyThread31 extends Thread{

    private Object lock;

    public MyThread31(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        ThreadDomain31 td = new ThreadDomain31();

        td.testMethod(lock);

    }

    public static void main(String[] args)

    {

        Object lock = new Object();

        MyThread31 mt0 = new MyThread31(lock);

        MyThread31 mt1 = new MyThread31(lock);

        mt0.start();

        mt1.start();

    }

}

运行结果如下

Thread-0 Begin wait()

Thread-1 Begin wait()

假如wait()不释放锁，我们应该执行完同步方法，我们应该打印

Thread-1 Begin wait()

Thread-1 End wait

Thread-0 Begin wait()

Thread-0 End wait

反之，证明了wait释放锁，放弃了同步块的执行权。

notify()不释放锁

ThreadDomain32类，用sleep为了说明此时没有其他线程进入同步代码块，即不释放锁。

public class ThreadDomain32 {

    public void testMethod(Object lock)

    {

        try

        {

            synchronized (lock)

            {

                System.out.println("Begin wait(), ThreadName = " + Thread.currentThread().getName());

                lock.wait();

                System.out.println("End wait(), ThreadName = " + Thread.currentThread().getName());

            }

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

    }

    public void synNotifyMethod(Object lock)

    {

        try

        {

            synchronized (lock)

            {

                System.out.println("Begin notify(), ThreadName = " + Thread.currentThread().getName());

                lock.notify();

                Thread.sleep(5000);

                System.out.println("End notify(), ThreadName = " + Thread.currentThread().getName());

            }

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

    }

}

MyThread32_0类

public class MyThread32_0 extends Thread{

    private Object lock;

    public MyThread32_0(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        ThreadDomain32 td = new ThreadDomain32();

        td.testMethod(lock);

    }

}

MyThread32_1类

public class MyThread32_1 extends Thread{

    private Object lock;

    public MyThread32_1(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        ThreadDomain32 td = new ThreadDomain32();

        td.synNotifyMethod(lock);

    }

}

输出结果如下

Begin wait(), ThreadName = Thread-0

Begin notify(), ThreadName = Thread-1

End notify(), ThreadName = Thread-1

End wait(), ThreadName = Thread-0

Begin notify(), ThreadName = Thread-2

End notify(), ThreadName = Thread-2

Begin wait()和End notify()紧密相连，我们已经使用了sleep(5000),5s之间足够让其他线程进入同步代码块了，说明notify的线程(mt1和mt2)没有释放锁。

notifyAll()唤醒所有线程

ThreadDomain34类

public class ThreadDomain34 {

    public void testMethod(Object lock)

    {

        try

        {

            synchronized (lock)

            {

                System.out.println("Begin wait(), ThreadName = " + Thread.currentThread().getName());

                lock.wait();

                System.out.println("End wait(), ThreadName = " + Thread.currentThread().getName());

            }

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

    }

}

MyThread34_0类

public class MyThread34_0 extends Thread {

    private Object lock;

    public MyThread34_0(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        ThreadDomain34 td = new ThreadDomain34();

        td.testMethod(lock);

    }

}

MyThread34_1类

public class MyThread34_1 extends Thread{

    private Object lock;

    public MyThread34_1(Object lock)

    {

        this.lock = lock;

    }

    public void run()

    {

        synchronized (lock)

        {

            lock.notifyAll();

        }

    }

}

main方法，启动三个线程并使其wait，启动mt3线程notifyAll

public class MyThread34_main {

    public static void main(String[] args) throws InterruptedException {

        Object lock = new Object();

        MyThread34_0 mt0 = new MyThread34_0(lock);

        MyThread34_0 mt1 = new MyThread34_0(lock);

        MyThread34_0 mt2 = new MyThread34_0(lock);

        mt0.start();

        mt1.start();

        mt2.start();

        Thread.sleep(1000);

        MyThread34_1 mt3 = new MyThread34_1(lock);

        mt3.start();

    }

}