Изучение многопоточности в Java (7) Расскажите о LockSupport.park() и LockSupport.unpark()

Пожалуйста, указывайте первоисточник при перепечатке, спасибо!

Блог HappyFeet

В последнее время я был занят поиском работы и дома, и у меня много дел, кроме того, я сменил город, поэтому у меня нет времени вести блог. Теперь, когда работа определена, а дом сдан, пришло время скорректировать свой менталитет и начать вести блог.

Честно говоря, я за это время познакомился со многими компаниями, много общался со многими интервьюерами, и у меня много чувств, но я еще не решил, как это написать. опыт в течение этого периода в сообщение в блоге как можно скорее.Выходите, пожалуйста, будьте терпеливы!

Отложив пока интервью, давайте сегодня поговорим об основных принципах LockSupport.park() и LockSupport.unpark().

Почему обсуждаются эти два метода? Причина в том, что при чтении исходного кода AQS я обнаружил, что эти два метода вызываются очень много! Поэтому, прежде чем продолжить углубленное изучение исходного кода AQS, давайте ознакомимся с базовой реализацией LockSupport.park() и LockSupport.unpark(), чтобы заложить основу для последующего изучения AQS.

1. LockSupport.park() и LockSupport.unpark()

1. Основы, которые вы должны знать

В переводе park на китайский язык означает «парковка», а значение этого метода в коде — «приостановка» текущего потока.

На самом деле в классе LockSupport предусмотрено множество методов, и наиболее часто используемые из них следующие:

park(): приостановить текущий поток на неопределенный срок.
parkNanos(long nanos): приостановить текущий поток на некоторое время
parkUntil(длинный срок): приостановить текущий поток до крайнего срока
unpark (нить потока): разбудить поток потока

Среди них unpark используется для пробуждения потока, а остальные три используются для его приостановки.

Приостановленные потоки делятся на неопределенные и конечные приостановки, а соответствующие состояния потока — WAITING (ожидание на неопределенный срок) и TIMED_WAITING (ожидание на неопределенный срок).

Есть три способа возобновить поток, который был приостановлен на неопределенный срок:

Другие потоки вызывают метод unpark, параметром является приостановленный поток.
Другой поток прерывает приостановленный поток
The call spuriously (that is, for no reason) returns.
По поводу ложного пробуждения можно обратиться к следующей информации:
- Why does pthread_cond_wait have spurious wakeups?
- Do spurious wakeups in Java actually happen?
Из-за наличия ложного пробуждения при вызове парковки обычно используется spin.Псевдокод выглядит следующим образом:
```
    while (!canProceed()) { 
        ...
        LockSupport.park(this);
    }
```

Помимо трех вышеперечисленных, есть еще четвертый способ приостановки на ограниченный срок:

Ожидаемый период времени прошел или указанный срок был достигнут

2. Каковы их характеристики?

This class associates, with each thread that uses it, a permit.

Java-документацияКак уже упоминалось, каждый поток связан с разрешением.

Когда лицензия доступна, вызов на парковку вернется немедленно, в противном случае он может быть приостановлен (ложное пробуждение не будет приостановлено); если лицензия недоступна, вызов на отпарку сделает лицензию доступной, а если сама лицензия доступен, вызов unpark не будет иметь никакого эффекта.

Это может быть не так ясно, если смотреть на текст напрямую, давайте рассмотрим несколько примеров:

Пример 1: park приостанавливает поток, unpark пробуждает приостановленный поток

public static void exampleOne() {
    Thread thread = new Thread(() -> {
        while (flag) {

        }

        System.out.println("before first park");
        LockSupport.park();
        System.out.println("after first park");
        LockSupport.park();
        System.out.println("after second park");

    });

    thread.start();

    flag = false;

    sleep(20);

    System.out.println("before unpark.");
    LockSupport.unpark(thread);
}

выходной результат

before first park
before unpark
after first park

анализировать

Во-первых, лицензия изначально недоступна, поэтому после вызова park поток приостанавливается, а последующий основной поток вызывает метод unpark, чтобы разбудить приостановленный поток, выводяafter first park, сразу после того, как поток вызывает парк и продолжает быть приостановленным.

Пример 2: эффект unpark не накапливается

private static void exampleTwo() {
    Thread thread = new Thread(() -> {
        while (flag) {

        }

        System.out.println("before first park");
        LockSupport.park();
        System.out.println("after first park");
        LockSupport.park();
        System.out.println("after second park");

    });

    thread.start();

    LockSupport.unpark(thread);
    LockSupport.unpark(thread);

    flag = false;
}

выходной результат

before first park
after first park

анализировать

Основной поток сначала выполняет две операции unpark над потоком, а затем поток дважды подряд вызывает метод park.Получается, что вторая парковка приостанавливает поток, это в основном отражает то, что эффект unpark не будет накапливаться. лицензия доступна, вызовите метод unpark не действует.

Пример 3: Влияние прерывания на парковку

private static void exampleThree() {
    Thread thread = new Thread(() -> {

        System.out.println("before first park");
        LockSupport.park();
        System.out.println("after first park");
        LockSupport.park();
        System.out.println("after second park");
        System.out.println("isInterrupted is " + Thread.interrupted());
        System.out.println("isInterrupted is " + Thread.interrupted());
        LockSupport.park();
        System.out.println("after third park");
    });

    thread.start();

    sleep(200);

    thread.interrupt();
}

выходной результат

before first park
after first park
after second park
isInterrupted is true
isInterrupted is false

анализировать

Всего поток трижды вызывал парк. Первые два вызова не отличаются друг от друга. Thread.interrupted() вызывается дважды перед третьим вызовом. третий Второй парк приостанавливает нить, почему?

Давайте сначала посмотрим на роль Thread.interrupted(): ** Определить прерванное состояние текущего потока и одновременно сбросить прерванное состояние. ** На самом деле вам нужно вызвать Thread.interrupted() только один раз здесь, и он вызывается дважды, чтобы проверить изменение статуса прерывания потока.

Когда состояние прерванного потока равно true, функция park теряет свое действие и не приостанавливает поток, а когда для сброса состояния прерывания вызывается Thread.interrupted(), функция park возобновляет свое действие.

Итак, вот выводы:Прерывание потока приведет к аннулированию парковки.

Полный пример портала кода

2. Отслеживание источника

Главное событие здесь! Выше было упомянуто несколько методов парковки и разпарковки, которые фактически соответствуют двум нативным методам UNSAFE.park(boolean isAbsolute, long time) и UNSAFE.unpark(Object thread).

Давайте посмотрим, как реализованы эти два нативных метода.

1. UNSAFE.park (логическое значение isAbsolute, долгое время)

(1)`unsafe.cpp#Unsafe_Park`:`openjdk/hotspot/src/share/vm/prims/unsafe.cpp`

UNSAFE_ENTRY(void, Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time))
  UnsafeWrapper("Unsafe_Park");
  EventThreadPark event;
#ifndef USDT2
  HS_DTRACE_PROBE3(hotspot, thread__park__begin, thread->parker(), (int) isAbsolute, time);
#else /* USDT2 */
   HOTSPOT_THREAD_PARK_BEGIN(
                             (uintptr_t) thread->parker(), (int) isAbsolute, time);
#endif /* USDT2 */
  JavaThreadParkedState jtps(thread, time != 0);
  thread->parker()->park(isAbsolute != 0, time);
#ifndef USDT2
  HS_DTRACE_PROBE1(hotspot, thread__park__end, thread->parker());
#else /* USDT2 */
  HOTSPOT_THREAD_PARK_END(
                          (uintptr_t) thread->parker());
#endif /* USDT2 */
  ...
UNSAFE_END

Нажмите, чтобы просмотреть исходный код

Здесь есть несколько ветвей суждений, но видно, что независимо от того, какая это ветвь, в конечном итоге она пойдет.park(bool isAbsolute, jlong time)Сюда.

(2)`os_linux.cpp#Parker::park`:`openjdk/hotspot/src/os/linux/vm/os_linux.cpp`

void Parker::park(bool isAbsolute, jlong time) {
			...
  if (Atomic::xchg(0, &_counter) > 0) return;

  Thread* thread = Thread::current();
  assert(thread->is_Java_thread(), "Must be JavaThread");
  JavaThread *jt = (JavaThread *)thread;

  		...
  if (Thread::is_interrupted(thread, false)) {
    return;
  }

  // Next, demultiplex/decode time arguments
  timespec absTime;
  if (time < 0 || (isAbsolute && time == 0) ) { // don't wait at all
    return;
  }
  if (time > 0) {
    unpackTime(&absTime, isAbsolute, time);
  }
			...
  if (Thread::is_interrupted(thread, false) || pthread_mutex_trylock(_mutex) != 0) {
    return;
  }
  
  int status ;
  if (_counter > 0)  { // no wait needed
    _counter = 0;
    status = pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
    // Paranoia to ensure our locked and lock-free paths interact
    // correctly with each other and Java-level accesses.
    OrderAccess::fence();
    return;
  }

  assert(_cur_index == -1, "invariant");
  if (time == 0) {
    _cur_index = REL_INDEX; // arbitrary choice when not timed
    status = pthread_cond_wait (&_cond[_cur_index], _mutex) ;
  } else {
    _cur_index = isAbsolute ? ABS_INDEX : REL_INDEX;
    status = os::Linux::safe_cond_timedwait (&_cond[_cur_index], _mutex, &absTime) ;
    if (status != 0 && WorkAroundNPTLTimedWaitHang) {
      pthread_cond_destroy (&_cond[_cur_index]) ;
      pthread_cond_init    (&_cond[_cur_index], isAbsolute ? NULL : os::Linux::condAttr());
    }
  }

  		...
  _counter = 0 ;
  status = pthread_mutex_unlock(_mutex) ;
  OrderAccess::fence();

  // If externally suspended while waiting, re-suspend
  if (jt->handle_special_suspend_equivalent_condition()) {
    jt->java_suspend_self();
  }
}

Нажмите, чтобы просмотреть исходный код

Во-первых, с помощью метода Atomic::xchg(0, &_counter) установите для _counter значение 0. Если исходный _counter > 0, исходная лицензия доступна и возвращается напрямую;
Если текущий поток находится в прерванном состоянии, возвращайтесь напрямую;

Thread::is_interrupted(thread, false) будет оценивать только прерванное состояние потока и не будет сбрасывать его прерванное состояние; Thread.interrupted() вызывает Thread::is_interrupted(thread, true) и прерванное состояние потока будет судить в то же время его сброс.

Если time
Оцените значение _counter еще раз, если оно больше 0, это означает, что исходная лицензия доступна, установите ее на 0, а затем вернитесь напрямую. Между Atomic::xchg и получением блокировки должна быть операция unpark, чтобы значение _counter стало равным 1;
Судя по тому, что time == 0, вызовите метод pthread_cond_wait для ожидания на неопределенный срок; в противном случае вызовите метод os::Linux::safe_cond_timedwait, который, наконец, вызовет pthread_cond_timedwait, тем самым входя в режим ожидания на неопределенный срок.

Для pthread_cond_wait вы можете прочитать эту статью:Подробное объяснение pthread_cond_wait

Он очень похож на механизм Object.wait, Object.notify и Object.notifyAll.

2. UNSAFE.unpark (Объектный поток)

(1)`unsafe.cpp#Unsafe_Unpark`:`openjdk/hotspot/src/share/vm/prims/unsafe.cpp`

UNSAFE_ENTRY(void, Unsafe_Unpark(JNIEnv *env, jobject unsafe, jobject jthread))
  UnsafeWrapper("Unsafe_Unpark");
  Parker* p = NULL;
  if (jthread != NULL) {
    oop java_thread = JNIHandles::resolve_non_null(jthread);
    if (java_thread != NULL) {
      jlong lp = java_lang_Thread::park_event(java_thread);
      if (lp != 0) {
        // This cast is OK even though the jlong might have been read
        // non-atomically on 32bit systems, since there, one word will
        // always be zero anyway and the value set is always the same
        p = (Parker*)addr_from_java(lp);
      } else {
        // Grab lock if apparently null or using older version of library
        MutexLocker mu(Threads_lock);
        java_thread = JNIHandles::resolve_non_null(jthread);
        if (java_thread != NULL) {
          JavaThread* thr = java_lang_Thread::thread(java_thread);
          if (thr != NULL) {
            p = thr->parker();
            if (p != NULL) { // Bind to Java thread for next time.
              java_lang_Thread::set_park_event(java_thread, addr_to_java(p));
            }
          }
        }
      }
    }
  }
  if (p != NULL) {
#ifndef USDT2
    HS_DTRACE_PROBE1(hotspot, thread__unpark, p);
#else /* USDT2 */
    HOTSPOT_THREAD_UNPARK(
                          (uintptr_t) p);
#endif /* USDT2 */
    p->unpark();
  }
UNSAFE_END

Нажмите, чтобы просмотреть исходный код

Предыдущий большой фрагмент кода присваивает значения Parker* p и, наконец, вызывает метод unpark p: p->unpark().

(2)`os_linux.cpp#Parker::unpark`:`openjdk/hotspot/src/os/linux/vm/os_linux.cpp`

void Parker::unpark() {
  int s, status ;
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant") ;
  s = _counter;
  _counter = 1;
  if (s < 1) {
    // thread might be parked
    if (_cur_index != -1) {
      // thread is definitely parked
      if (WorkAroundNPTLTimedWaitHang) {
        status = pthread_cond_signal (&_cond[_cur_index]);
        assert (status == 0, "invariant");
        status = pthread_mutex_unlock(_mutex);
        assert (status == 0, "invariant");
      } else {
        status = pthread_mutex_unlock(_mutex);
        assert (status == 0, "invariant");
        status = pthread_cond_signal (&_cond[_cur_index]);
        assert (status == 0, "invariant");
      }
    } else {
      pthread_mutex_unlock(_mutex);
      assert (status == 0, "invariant") ;
    }
  } else {
    pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
  }
}

Нажмите, чтобы просмотреть исходный код

Метод unpark на самом деле очень прост.Сначала получите блокировку с помощью pthread_mutex_lock, затем установите _counter в 1, а затем определите, приостановлен ли поток в данный момент.Если это так, разбудите приостановленный поток с помощью pthread_cond_signal, а затем снимите блокировку.

3. Заключение

Изучение знаний представляет собой цепочку Когда вы изучаете элемент знания, которого не знаете, легко встретить новые элементы знания, которых вы не знаете, а затем вы продолжаете контактировать с новыми элементами знаний и изучать новый контент. поймите новые точки знаний, и постепенно вы обнаружите, что идеи многих точек знаний на самом деле одинаковы, и их не так уж сложно изучить.

Например, когда вы увидите механизм pthread_cond_wait, вы подумаете об Object.wait, потому что в их реализации много общего, и его относительно просто понять.

Напоследок оставлю вопрос, можно подумать: в чем разница между LockSupport.park и Object.wait?

Использованная литература:

(1)Углубленный анализ interrupt() и LockSupport.park(), вызванный несколькими небольшими случаями

(2)Class LockSupport

1. LockSupport.park() и LockSupport.unpark()

1. Основы, которые вы должны знать

2. Каковы их характеристики?

Пример 1: park приостанавливает поток, unpark пробуждает приостановленный поток

Пример 2: эффект unpark не накапливается

Пример 3: Влияние прерывания на парковку

2. Отслеживание источника

1. UNSAFE.park (логическое значение isAbsolute, долгое время)

(1)unsafe.cpp#Unsafe_Park:openjdk/hotspot/src/share/vm/prims/unsafe.cpp

(2)os_linux.cpp#Parker::park:openjdk/hotspot/src/os/linux/vm/os_linux.cpp

2. UNSAFE.unpark (Объектный поток)

(1)unsafe.cpp#Unsafe_Unpark:openjdk/hotspot/src/share/vm/prims/unsafe.cpp

(2)os_linux.cpp#Parker::unpark:openjdk/hotspot/src/os/linux/vm/os_linux.cpp

3. Заключение

(1)`unsafe.cpp#Unsafe_Park`:`openjdk/hotspot/src/share/vm/prims/unsafe.cpp`

(2)`os_linux.cpp#Parker::park`:`openjdk/hotspot/src/os/linux/vm/os_linux.cpp`

(1)`unsafe.cpp#Unsafe_Unpark`:`openjdk/hotspot/src/share/vm/prims/unsafe.cpp`

(2)`os_linux.cpp#Parker::unpark`:`openjdk/hotspot/src/os/linux/vm/os_linux.cpp`