初识 Java 并发

并发

1、线程的生命周期

public enum State {
        
        NEW,
    
        RUNNABLE,
    
        BLOCKED,

        WAITING,
    
        TIMED_WAITING,

        TERMINATED;
    }

在 Thread 类中有一枚举类 State 定义了线程的六种状态

• NEW状态：新创建出来一个线程
• RUNNABLE：调用 start() 方法之后，会处于此状态，等待获取时间片之后进行 RUNNING 执行 run()
• BLOCK：当我们有两个 a,b 去调用 synchronized add() 方法时，若 a 先获得锁，b 在去获得锁时，便会返回 BLOCK 状态
• WAITING：处于一个等待状态，调用 wait,join,LockSupport.park 方法后，会处于此状态
• TIME_WAITING：和 WAITING 的区别就是，这个可以设置超时时间，等待超过超时时间后，会进行释放
• TERMINATED：线程执行完毕，或者出现异常

2、线程状态的演示

// WAITING 状态
new Thread(() -> {
            while (true) {
                synchronized (ThreadStatus.class) {
                    try {
                        ThreadStatus.class.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, "wait状态").start();

// TIME_WAITING 状态
new Thread(() -> {
            while (true) {
                synchronized (ThreadStatus.class) {
                    try {
                        TimeUnit.SECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, "TIME_WAITING状态").start();

// BLOCK 状态
App app = new App();
new Thread(() -> app.demo()).start();
new Thread(() -> app.demo(), "BLOCK状态").start();

static class App {
        public synchronized void demo() {
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

可以通过 jps 查看 pid，通过 jstack pid 查看jvm 相关信息

"BLOCK状态" #15 prio=5 os_prio=0 tid=0x00000000207c9000 nid=0x9ca4 waiting for monitor entry [0x000000002132f000]
   java.lang.Thread.State: BLOCKED (on object monitor)
        at ThreadStatus$App.demo(ThreadStatus.java:47)
        - waiting to lock <0x000000076bebe398> (a ThreadStatus$App)
        at ThreadStatus.lambda$main$3(ThreadStatus.java:40)
        at ThreadStatus$$Lambda$4/558638686.run(Unknown Source)
        at java.lang.Thread.run(Thread.java:748)

"TIME_WAITING状态" #13 prio=5 os_prio=0 tid=0x00000000207c4800 nid=0xa6ec waiting on condition [0x000000002112e000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
        at java.lang.Thread.sleep(Native Method)
        at java.lang.Thread.sleep(Thread.java:340)
        at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
        at ThreadStatus.lambda$main$1(ThreadStatus.java:29)
        - locked <0x000000076bb9a968> (a java.lang.Class for ThreadStatus)
        at ThreadStatus$$Lambda$2/1078694789.run(Unknown Source)
        at java.lang.Thread.run(Thread.java:748)

"wait状态" #12 prio=5 os_prio=0 tid=0x00000000207bf800 nid=0x56bc in Object.wait() [0x000000002102f000]
   java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        - waiting on <0x000000076bb9a968> (a java.lang.Class for ThreadStatus)
        at java.lang.Object.wait(Object.java:502)
        at ThreadStatus.lambda$main$0(ThreadStatus.java:17)
        - locked <0x000000076bb9a968> (a java.lang.Class for ThreadStatus)
        at ThreadStatus$$Lambda$1/1324119927.run(Unknown Source)
        at java.lang.Thread.run(Thread.java:748)

3、线程是怎么启动的

public synchronized void start() {

        if (threadStatus != 0)
            throw new IllegalThreadStateException();

        group.add(this);

        boolean started = false;
        try {
            start0();
            started = true;
        } finally {
            try {
                if (!started) {
                    group.threadStartFailed(this);
                }
            } catch (Throwable ignore) {
              
            }
        }
    }

当我们调用 start 方法后，由源码可见，在其内部调用了 syary0() 的一个 native 方法

在 openJDK 的 Thread.c 中可以看到

​ 也就是说，start0 这个方法，调用的是 JVM_StartThread 这个方法

// JVM.CPP
JVM_ENTRY(void, JVM_StartThread(JNIEnv* env, jobject jthread))
  JVMWrapper("JVM_StartThread");
  JavaThread *native_thread = NULL;

      size_t sz = size > 0 ? (size_t) size : 0;
      native_thread = new JavaThread(&thread_entry, sz);
 		... ...

  	Thread::start(native_thread);

JVM_END

JVM_ENTRY 就是定义了 JVM_StartThread 这个函数，可以发现，在代码中进行了 new JavaThread 的操作，并在之后调用了 Thread 的 start 方法，将 new 的 Thread 的传入，启动这个线程

new JavaThread 的相关操作在代码 Thread.cpp 文件下

// Thread.cpp
JavaThread::JavaThread(ThreadFunction entry_point, size_t stack_sz) :
  Thread()
#if INCLUDE_ALL_GCS
  , _satb_mark_queue(&_satb_mark_queue_set),
  _dirty_card_queue(&_dirty_card_queue_set)
#endif // INCLUDE_ALL_GCS
{
  if (TraceThreadEvents) {
    tty->print_cr("creating thread %p", this);
  }
  initialize();
  _jni_attach_state = _not_attaching_via_jni;
  set_entry_point(entry_point);
  
  // 通过判断类型
  os::ThreadType thr_type = os::java_thread;
  thr_type = entry_point == &compiler_thread_entry ? os::compiler_thread :
                                                     os::java_thread;
  // 操作系统 底层创建 java_thread
   os::create_thread(this, thr_type, stack_sz);
  _safepoint_visible = false;
 	... ...
}

在注释的地方，可以看到 new JavaThread 主要是通过操作系统底层创建出来一个 Thread 线程

接着就是 start 启动，同样也是通过操作系统底层对线程进行启动

// Thread.cpp
// Thread.start()
void Thread::start(Thread* thread) {
  trace("start", thread);
  // Start is different from resume in that its safety is guaranteed by context or
  // being called from a Java method synchronized on the Thread object.
  if (!DisableStartThread) {
    if (thread->is_Java_thread()) {
      // Initialize the thread state to RUNNABLE before starting this thread.
      // Can not set it after the thread started because we do not know the
      // exact thread state at that time. It could be in MONITOR_WAIT or
      // in SLEEPING or some other state.
      java_lang_Thread::set_thread_status(((JavaThread*)thread)->threadObj(),
                                          java_lang_Thread::RUNNABLE);
    }
    // 通过 操作系统底层，启动线程
    os::start_thread(thread);
  }
}

// os_Linux.cpp
static void *java_start(Thread *thread) {
 
  static int counter = 0;
  int pid = os::current_process_id();
  alloca(((pid ^ counter++) & 7) * 128);

  ThreadLocalStorage::set_thread(thread);

  OSThread* osthread = thread->osthread();
  Monitor* sync = osthread->startThread_lock();

 ... ...

  // handshaking with parent thread
  {
    MutexLockerEx ml(sync, Mutex::_no_safepoint_check_flag);

    // notify parent thread
    osthread->set_state(INITIALIZED);
    sync->notify_all();

    // wait until os::start_thread()
    while (osthread->get_state() == INITIALIZED) {
      sync->wait(Mutex::_no_safepoint_check_flag);
    }
  }

  // 进行回调 run() 方法
  thread->run();

  return 0;
}

由以上也就可以得出为什么线程启动的方法，不是调用 run 方法，而是调用 start 方法，去启动线程，同时，在底层代码中，会对 run 方法进行回调 ,这也就是要重写 run 方法的原因

4、线程的终止

4.1 interrupt 方法停止

	public static int i = 0;
    public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                i++;
            }
            System.err.println(i);
        });
        thread.start();
        TimeUnit.SECONDS.sleep(1);
        thread.interrupt();
    }

isInterrupted方法，默认是 false，interrupt方法，设置值为 true

找到源码，发现调用的是 Thread 里面的 interrupt ，接着往下找

// JVM.CPP
JVM_ENTRY(void, JVM_Interrupt(JNIEnv* env, jobject jthread))
  JVMWrapper("JVM_Interrupt");
	... ...
  JavaThread* thr = java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread));
  if (thr != NULL) {
    // 调用 Thread 里面的 interrupt
    Thread::interrupt(thr);
  }
JVM_END

// Thread.cpp
void Thread::interrupt(Thread* thread) {
  trace("interrupt", thread);
  debug_only(check_for_dangling_thread_pointer(thread);)
  // 通过底层操作系统，进行操作
  os::interrupt(thread);
}

// os_Linux.cpp
void os::interrupt(Thread* thread) {
  assert(Thread::current() == thread || Threads_lock->owned_by_self(),
    "possibility of dangling Thread pointer");
	// 获取本地线程对象
  OSThread* osthread = thread->osthread();

  if (!osthread->interrupted()) { // interrupted 默认值为 flase
    // 设置成 true
    osthread->set_interrupted(true);
      // 设置内存屏障，目的是使得 interrupted 状态对其他线程立即可见
    OrderAccess::fence();
    // SleepEvent 相当于 Thread.sleep ,若处于，则通过 unpark 进行解锁
    ParkEvent * const slp = thread->_SleepEvent ;
    if (slp != NULL) slp->unpark() ;
  }

  // For JSR166. Unpark even if interrupt status already was set
  if (thread->is_Java_thread())
    ((JavaThread*)thread)->parker()->unpark();
  // 相当于 sync 代码块中 object.wait 方法，调用 unpark 进行解锁
  ParkEvent * ev = thread->_ParkEvent ;
  if (ev != NULL) ev->unpark() ;
}

// opThread.hpp
volatile jint _interrupted; // 定义了一个 标记

// 调用此方法，进行标记的 赋值
void set_interrupted(bool z)                      { _interrupted = z ? 1 : 0; }

通过调用 interrupt 方法，会将 interrupted 属性进行赋值，并将进入 WAITING 状态的进行释放

通过这种方法，设置中断标记，优雅中断，并不是直接武断将线程停止，使线程在终止时有时间去清理资源等

4.2 interrupted 方法复位停止

​ interrupted 方法，对中断标记进行复位（恢复默认值）

Thread thread = new Thread(() -> {
            if (!Thread.currentThread().isInterrupted()) {
                System.err.println("线程当前状态: " + 				                					Thread.currentThread().isInterrupted());
                Thread.currentThread().interrupted();
                System.err.println("线程复位之后的状态: " + 				                				Thread.currentThread().isInterrupted());
            }
        });
        thread.start();
        TimeUnit.SECONDS.sleep(1);
        thread.interrupt();
线程当前状态: false
线程复位之后的状态: false

// JVM.CPP
JVM_QUICK_ENTRY(jboolean, JVM_IsInterrupted(JNIEnv* env, jobject jthread, jboolean clear_interrupted))
  JVMWrapper("JVM_IsInterrupted");

  ... ...
  oop java_thread = JNIHandles::resolve_non_null(jthread);
  MutexLockerEx ml(thread->threadObj() == java_thread ? NULL : Threads_lock);
  ... ...
  JavaThread* thr = java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread));
  if (thr == NULL) {
    return JNI_FALSE;
  } else {
      // 调用了 Thread 中的 is_interrupted
    return (jboolean) Thread::is_interrupted(thr, clear_interrupted != 0);
  }
JVM_END

// Thread.cpp
bool Thread::is_interrupted(Thread* thread, bool clear_interrupted) {
  trace("is_interrupted", thread);
  debug_only(check_for_dangling_thread_pointer(thread);)
  // 调用底层操作
  return os::is_interrupted(thread, clear_interrupted);
}

// os_Linux.cpp
bool os::is_interrupted(Thread* thread, bool clear_interrupted) {
  assert(Thread::current() == thread || Threads_lock->owned_by_self(),
    "possibility of dangling Thread pointer");

  OSThread* osthread = thread->osthread();
	// 获得线程的中断状态
  bool interrupted = osthread->interrupted();

  // 对标记设置为 false ,恢复为默认值
  if (interrupted && clear_interrupted) {
    osthread->set_interrupted(false);
    // consider thread->_SleepEvent->reset() ... optional optimization
  }

  return interrupted;
}

进行复位，主要就是要告诉外界，我已经收到你的中断信号，但具体什么时候中断是由线程自己做决定。

4.3 阻塞方法抛出异常

		Thread thread = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    System.err.println("收到信号");
                    break;
                }
            }
            System.err.println(i);
        });
        thread.start();
        TimeUnit.SECONDS.sleep(1);
        thread.interrupt();

像 object.wait, object.sleep 的阻塞方法，都会抛出一个 InterruptedException 异常，通过对这个异常进行处理，达到终止线程的目的

转载：https://blog.csdn.net/qq_40293674/article/details/104699305