多线程编程的精华：探索 Java 中的多种线程开启方式

聊到多线程,我们优先谈一下什么是进程?什么又是线程呢?教科书式的说法是:进程是计算机操作系统进行内存分配的最小单元;线程是计算机操作系统进行任务分配的最小单元

咱们用一个小故事,讲一下什么是线程?什么是进程?

我们可以将你们村比作一个操作系统,每家每户都可以比作一个应用程序,村长家比作QQ,二狗家比作360,胖丫家比作Office

这样我们就可以理解为:每家每户其实是运行在操作系统中的程序,这里的每个程序都是单独的一个进程,每个进程相互之间都是独门独户,平日里风平浪静,偶尔也会出现村长和二狗家干仗的情况.

每个家庭中的每一个人,都可以理解为不同的线程,每个线程可能干相同的事,也可能干不同的事情,但是他们可以共用一部分区域(内存),比如看电视,上厕所,这些都是公用区域.

当然也会出现线程之间抢资源的情况,有一天,姐姐和弟弟有一天心血来潮突然都想看电视,弟弟想看蜡笔小新,姐姐想看小猪佩奇,恰巧他爸多买了一个遥控器,弟弟选择CCTV-少儿频道,姐姐选择山东少儿频道,频道换来换去,电视一会是这个一会是那个,这就是所说的多个线程同时写同一个数据,且读到的数据也不相同,这样线程是不安全的.

恰巧他老爹有一天通过偷拍摄像头看到了这一幕,于是决定,将其中一个遥控器给藏起来,并对姐弟俩说,以后遥控器谁先拿到谁先看,没抢到的那个后面等着,谁在抢来抢去打屁屁,这就相当于加了一把synchronized锁,第一线程未释放锁之前,后面的线程只能等到第一个线程将锁释放后才能持有.

故事先讲到这里,今天主要是聊的是如何开启线程,后面再讲如何控制线程安全,以及公平锁非公平锁,递归锁等

如果你是window电脑,可以打开任务管理器,可以看到”进程”一列.如图,可以通过状态识别当前进程的执行状态,CPU使用率,内存的使用情况,对磁盘的IO使用以及当前进程对网络的的吞吐量等,这里就是我们的进程,由计算机为其分配的内存,如果你本地运行这Java程序,或是启动着Tomcat可以来这里看一下你内存的使用情况.

如果是Linux或者是Mac,则可以通过top命令查看当前运行中的进程,如图

继承Thread,重写run方法

那接下来我们讲一下如何启动线程,启动线程的方式有很多,比较传统的方式也就那两种,继承Thread,重写run方法,调用start方法;实现Runnable接口,实现run方法,我们先看下代码,比较传统的的方式也得看一下,继承Thread,重写run()方法,调用start方法,这里尤其注意,是调用start方法,而不是重写的run方法

/**
* desc:集成Thread类,重写run方法,调用start方法
* created by cuiyongxu on 2021/12/15 12:33 上午
*/
public class ThreadTask {public static void main(String[] args) {System.out.println(ThreadTask.class.getName() + "-" + Thread.currentThread().getId());ThreadItem threadItem = new ThreadItem();threadItem.start();}
}

class ThreadItem extends Thread {@Overridepublic void run() {System.out.println(this.getClass().getName() + "-" + Thread.currentThread().getId());}
}//执行结果:
//ThreadTask-1
//ThreadItem-10//如果将threadItem.start()换做为thradItem.run() 执行结果为
//ThreadTask-1
//ThreadItem-1

这时候有的同学会问了,为什么重写了Thread类的run方法,却要调用start来启动线程呢?那我们可以进入start方法中看下实际上调用的是哪个方法,通过以下截图可以看到,其实调用的是start0方法,图中也将start0方法进行了标记,其实是一个native方法,那start0种具体做了什么事情呢? 那我们需要看下jvm底层的源码

感兴趣的同学可以去github上面搜一下openjdk的源码,以下片段截取于:Thread.c,我们能看到基于Thread的native 方法还不少呢,我们现在讲下start0至于其他的native方法,有兴趣的同学可以根据我们分析start0的方式去分析其他的方法.

通过下面的截图我们可以看到 start0是一个不需要入参且不需要返回值的一个方法,我们看下JVM_StartThread中的逻辑实现,这个方法在jvm.cpp中,

以下代码截取自jvm.cpp 2867行 (截止2021年12月16日)

JVM_ENTRY(void, JVM_StartThread(JNIEnv* env, jobject jthread))JavaThread *native_thread = NULL;// We cannot hold the Threads_lock when we throw an exception,// due to rank ordering issues. Example:  we might need to grab the// Heap_lock while we construct the exception.bool throw_illegal_thread_state = false;// We must release the Threads_lock before we can post a jvmti event// in Thread::start.{// Ensure that the C++ Thread and OSThread structures aren't freed before// we operate.MutexLocker mu(Threads_lock);// Since JDK 5 the java.lang.Thread threadStatus is used to prevent// re-starting an already started thread, so we should usually find// that the JavaThread is null. However for a JNI attached thread// there is a small window between the Thread object being created// (with its JavaThread set) and the update to its threadStatus, so we// have to check for thisif (java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread)) != NULL) {throw_illegal_thread_state = true;} else {// We could also check the stillborn flag to see if this thread was already stopped, but// for historical reasons we let the thread detect that itself when it starts runningjlong size =java_lang_Thread::stackSize(JNIHandles::resolve_non_null(jthread));// Allocate the C++ Thread structure and create the native thread.  The// stack size retrieved from java is 64-bit signed, but the constructor takes// size_t (an unsigned type), which may be 32 or 64-bit depending on the platform.//  - Avoid truncating on 32-bit platforms if size is greater than UINT_MAX.//  - Avoid passing negative values which would result in really large stacks.NOT_LP64(if (size > SIZE_MAX) size = SIZE_MAX;)size_t sz = size > 0 ? (size_t) size : 0;// 看我中文是不是特别醒目  注意thread_entry的使用,看后面的代码片段native_thread = new JavaThread(&thread_entry, sz);// At this point it may be possible that no osthread was created for the// JavaThread due to lack of memory. Check for this situation and throw// an exception if necessary. Eventually we may want to change this so// that we only grab the lock if the thread was created successfully -// then we can also do this check and throw the exception in the// JavaThread constructor.if (native_thread->osthread() != NULL) {// Note: the current thread is not being used within "prepare".native_thread->prepare(jthread);}}}if (throw_illegal_thread_state) {THROW(vmSymbols::java_lang_IllegalThreadStateException());}assert(native_thread != NULL, "Starting null thread?");if (native_thread->osthread() == NULL) {ResourceMark rm(thread);log_warning(os, thread)("Failed to start the native thread for java.lang.Thread \"%s\"",JavaThread::name_for(JNIHandles::resolve_non_null(jthread)));// No one should hold a reference to the 'native_thread'.native_thread->smr_delete();if (JvmtiExport::should_post_resource_exhausted()) {JvmtiExport::post_resource_exhausted(JVMTI_RESOURCE_EXHAUSTED_OOM_ERROR | JVMTI_RESOURCE_EXHAUSTED_THREADS,os::native_thread_creation_failed_msg());}THROW_MSG(vmSymbols::java_lang_OutOfMemoryError(),os::native_thread_creation_failed_msg());}#if INCLUDE_JFRif (Jfr::is_recording() && EventThreadStart::is_enabled() &&EventThreadStart::is_stacktrace_enabled()) {JfrThreadLocal* tl = native_thread->jfr_thread_local();// skip Thread.start() and Thread.start0()tl->set_cached_stack_trace_id(JfrStackTraceRepository::record(thread, 2));}
#endifThread::start(native_thread);JVM_END

注意这里的run_method_name ,这说明还有一部分代码需要贴出来,篇幅有点大了,有的没的说了一堆,这里注意下 run_method_name 实际已经不再jvm.cpp中了,而是在vmSymbols.hpp中,后面的不做细讲了,可能有的同学对C语言忘得差不多了,我们姑且跳过这个片段,继续聊我们Java相关的内容

static void thread_entry(JavaThread* thread, TRAPS) {HandleMark hm(THREAD);Handle obj(THREAD, thread->threadObj());JavaValue result(T_VOID);JavaCalls::call_virtual(&result,obj,vmClasses::Thread_klass(),vmSymbols::run_method_name(),vmSymbols::void_method_signature(),THREAD);
}

实现Runnable接口,实现run()方法

/**
* desc:
* created by cuiyongxu on 2021/12/15 12:34 上午
*/
public class RunnableTask {public static void main(String[] args) {System.out.println(RunnableTask.class.getName() + "-" + Thread.currentThread().getId());new Thread(new RunnableItem()).start();}
}class RunnableItem implements Runnable {@Overridepublic void run() {System.out.println(this.getClass().getName() + "-" + Thread.currentThread().getId());}
}//输出结果为:
//RunnableTask-1
//RunnableItem-10

有的同学就要问了 Thread和Runnable到底有什么区别,目前鄙人经验之谈,两者其实没有本质的区别,一个是接口,一个是实现类,并且实现类对接口做了更多功能的扩充,要是有人较真还是有区别的,那估么可以聊下Thread是类,只能单继承;Runnable是接口,可以多实现吧,但是真正生产环境下,一般很少有这样用的,我见过的唯一这样使用的是我的世界源码,如图.如果有时候的不对的,各位看官可赐教

实现Callable接口,实现call方法

通过FeatureTask启动创建一个线程,就能获取到线程执行的返回值,当然编写代码的方式有很多,以下编码方式只是其中一种


import com.google.common.collect.Lists;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
/**
* desc:
* created by cuiyongxu on 2021/12/15 12:36 上午
*/
public class CallableTask {public static void main(String[] args) throws Exception {System.out.println(CallableTask.class.getName() + "-" + Thread.currentThread().getId());FutureTask<List<String>> listFutureTask = new FutureTask<>(new CallableItem());Thread thread = new Thread(listFutureTask);thread.start();System.out.println(listFutureTask.get());}
}
class CallableItem implements Callable<List<String>> {@Overridepublic List<String> call() throws Exception {String msg = this.getClass().getName() + "-" + Thread.currentThread().getId();return Lists.newArrayList(msg);}
}
//返回结果为:
//CallableTask-1
//[CallableItem-10]

Executors.newCachedThreadPool

创建一个可缓存线程池,如果以前构建的线程可复用,则直接复用之前创建的线程了;如果线程不可复用,则会根据需要,创建新的线程,并将它添加到线程池中,如果线程在60s内未被使用,则会被终止并从线程池中移除;此种方式线程池个数无上限,虽然这么说,也是有最大限度的,但理论上不会达到,即最大线程数为: 2^31-1. 即Interger.MAX_VALUE,如图

使用此种线程池需要注意本身操作系统user processes最大进程值我mac 默认2784 ;可以通过 sudo ulimit -a 查询

如果使用线程池不当,则会报出异常: Exception in thread "main" java.lang.OutOfMemoryError: unable to create new native thread,例如以下程序:

@Test
public void cachedThreadPool() {ExecutorService executorService = Executors.newCachedThreadPool();for (int i = 0; i < 10000; i++) {executorService.submit(() -> {try {Thread.sleep(1000 * 5);System.out.println(Thread.currentThread().getId());} catch (InterruptedException e) {e.printStackTrace();}});}executorService.shutdown();
}

Executors.newFixedThreadPool(2)

创建一个定长线程池,不管在何时,最多可存在n个活动的线程数,当线程池中线程处于激活状态,则后续提交的任务,将被迫处于等待状态,如果线程中的线程执行过程中由于故障被终止,则会创建一个新的线程来替代它,池中的线程会一直存在

示例代码如下：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/**
* desc:
* created by cuiyongxu on 2021/12/16 10:45 下午
*/
public class NewFixedThreadPoolTask {public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(2);long startTime = System.currentTimeMillis();for (int i = 0; i < 10; i++) {int finalI = i;executorService.submit(() -> {System.out.println(finalI+"|"+(System.currentTimeMillis()-startTime));try {Thread.sleep(2000);} catch (InterruptedException e) {}});}executorService.shutdown();}
}/*
0|113
1|113
2|2119
3|2119
4|4120
5|4120
6|6121
7|6121
8|8125
9|8125*/

以上实例中设置最大线程数为2,在执行过程中,每个线程堵塞2s,stateTime只有第一次会初始化,故没两个相邻的线程打印出的耗时是完全相同的,以此证明,在同一时间内,最多只有两个线程可用.输出结果如下:

Executors.newScheduledThreadPool(2)

创建一个定长的线程池,该线程池支持延迟及周期性执行,以下代码为延迟执行,3s后开始执行


import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;/**
* desc:
* created by cuiyongxu on 2021/12/16 11:41 下午
*/
public class NewScheduledThreadPoolTask {public static void main(String[] args) {ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);long startTime = System.currentTimeMillis();for (int i = 0; i < 10; i++) {int finalI = i;executorService.schedule(() ->System.out.println(finalI + "|" + (System.currentTimeMillis() - startTime)), 3, TimeUnit.SECONDS);}executorService.shutdown();}
}

以下代码逻辑为,延迟0秒后开始执行,每3s执行一次

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;/**
* desc:
* created by cuiyongxu on 2021/12/16 11:41 下午
*/
public class NewScheduledThreadPoolTask {public static void main(String[] args) {ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);long startTime = System.currentTimeMillis();for (int i = 0; i < 10; i++) {int finalI = i;executorService.scheduleAtFixedRate(() ->System.out.println(finalI + "|" + (System.currentTimeMillis() - startTime)), 0, 3, TimeUnit.SECONDS);}}
}

Executors.newSingleThreadExecutor()

单一的线程池，该线程池中每时每刻只有一个线程能运行。后续线程必须等待当前执行的线程执行完了后，才能执行,需要遵循(FIFO),示例代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/**
* @Description :
* @Author : cuiyongxu
* @Date : 2021/12/19-12:58 上午
**/
public class NewSingleThreadExecutorTask {public static void main(String[] args) {ExecutorService executorService = Executors.newSingleThreadExecutor();for (int i = 0; i < 10; i++) {int finalI = i;executorService.execute(() -> {try {System.out.println(finalI);Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}});}}
}
/*
输出结果为:
0
1
2
3
4
5
6
7
8
9
*/

Executors.newWorkStealingPool()

优先说一下,WorkStealingPool是JDK8中新引进的一种并发线程池,它同以上4中通过Executors创建出来的线程池有所不同,以上4中线程池是通过对ThreadPoolExecutor的初始化扩展,而WorkStealingPool则是对ForkJoinPool的扩展,源码如下:

可以通过以上代码可以看出,默认情况下,WorkStealingPool最大线程数与当前java虚拟机可用的处理器数,且在执行过程中,是无法保证执行顺序的,例如文件下载功能,多个线程同时下载的场景,示例代码如下:


import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/**
* @Description :
* @Author : cuiyongxu
* @Date : 2021/12/19-12:58 上午
**/
public class NewSingleThreadExecutorTask {public static void main(String[] args) {ExecutorService executorService = Executors.newSingleThreadExecutor();for (int i = 0; i < 10; i++) {int finalI = i;executorService.execute(() -> {try {System.out.println(finalI);Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}});}}
}/*
执行结果:
0
2
1
3
4
5
8
9
6
7
*/

期待您的关注,欢迎访问我的博客