区块链技术博客
www.b2bchain.cn

Java源码

这篇文章主要介绍了Java源码,通过具体代码讲解6051并且分析了Java源码的详细步骤与相关技巧,需要的朋友可以参考下

本文实例讲述了Java源码。分享给大家供大家参考文章查询地址https://www.b2bchain.cn/?p=6051。具体如下:

Java集合源码

Java集合源码

发布日期:   2020-03-01
文章字数:   2.5k
阅读时长:   9 分
阅读次数:  

线程池源码

整体架构

Java源码

首先来了一个任务,先判断下核心线程池里的线程数有没有达到设定的值,如果没有,就可以新建个Worker出来处理任务,处理任务的时候如果这个任务为空,那这个worker就阻塞住,因为这个任务不需要处理,如果这个任务不为空,就会准备执行任务,这个Worker实现了Runnable接口,并且继承了AQS,那么它就会重写run方法,它重写的run方法里面调用了runWorker(),这个是真正执行任务的,执行任务时,那么首先就会所在执行任务前看看你有没有重写beforeExecute,如果重写了,那就会执行下,然后再执行任务,执行完也会看下有没有重写afterExecute。

如果来了一个任务,队列没满,就尝试放进阻塞队列,在尝试放入阻塞队列的时候,如果线程池状态异常,那就会拒绝任务,如果队列满了,那就要看此时你设置的maxinumPoolSize,看线程池里的线程有没有达到这个最大值,如果没有,那就可以创建线程出来执行任务,如果达到了,没有线程了,那就只能采取拒绝策略,常用的拒绝策略是 abortpolicy(抛异常),还有discardPolicy(直接丢弃任务),还有discardOldestPolicy(直接丢弃队列中最老的任务),还有个忘了

1、Executor

这个是多线程的顶级接口,只有一个execute方法

2、ExecutorService

Executor 的功能太弱,只有一个execute,ExecutorService接口 丰富了对任务的执行和管理的功能,它继承了Executor接口

// 线程池关闭,不会接受新的任务,也不会等待未完成的任务 void shutdown();  // executor 是否已经关闭了,返回值 true 表示已关闭 boolean isShutdown();  // 所有的任务是否都已经终止,是的话,返回 true boolean isTerminated();  // 在超时时间内,等待剩余的任务终止 boolean awaitTermination(long timeout, TimeUnit unit)  // 提交有返回值的任务,使用 get 方法可以阻塞等待任务的执行结果返回 <T> Future<T> submit(Callable<T> task);  //传入Runnable就没有返回结果了 Future<?> submit(Runnable task);  //还有其他一些方法,这里只是举例

3、AbstractExecutorService

这个是个一个抽象类,封装ExecutorService里的功能

4、注释得到的信息

  • 线程池可以使用Executors进行配置
    • Executors.newCachedThreadPool 无界的线程池,并且可以自动回收;
    • Executors#newFixedThreadPool 固定大小线程池;
    • Executors#newSingleThreadExecutor 单个线程的线程池;
  • 线程池提供可供扩展的参数设置,corsSize和maxSize,一般来说这两个值初始化时候就设定了,当然我们可以自己set,来修改这两个值
  • 线程池创建时有多种队列可供选择,有四种队列,可见我另一篇博客
    • LinkedBlockingQueue,无界阻塞队列
    • ArrayBlockingQueue,有界阻塞队列
    • DelayBlockingQueue,延迟队列
    • SynchronousQueue,同步阻塞队列
  • 线程池用了模版方法模式,提供了很多hook函数,比如beforeExecute()、afterExecute(),在线程执行前后需要做什么
  • 还有很多信息就不写了

5、ThreadPoolExecutor的重要属性

  • ctl:线程池状态字段,它是由两部分构成的
    • workerCount,wc 工作线程数, workerCount 最大到(2^29)-1,大概 5 亿个线程
    • runState,rs 线程池的状态
  • RUNNING:值为-536870912,表示线程池正在接受新任务或者处理队列里的任务
  • SHUTDOWN:值为0,表示不接受新任务,但仍在处理已经在队列里面的任务
  • STOP:值为-536870912,不接受新任务,也不处理队列中的任务,有正在执行的任务就中断
  • TIDYING:值为1073741824,它是整理状态,所有任务都被中断,此时wc是0
  • TERMINATED:terminated() 已经完成的时候
  • ThreadFactory threadFactory:使用 threadFactory 创建 thread
  • RejectedExecutionHandler handler:拒绝任务的 handler 处理类
  • corePoolSize:线程池线程核心数
  • maximumPoolSize:线程池最大线程数

Worker是线程池中运行任务的最小单元,Worker实现了Runnable接口,继承了AQS类。Worker本身就是一个Runnable,在它的构造方法中,getThreadFactory().newThread(this)这是话是说它是把自己作为任务传递给了thread

Worker(Runnable firstTask) {   setState(-1); // inhibit interrupts until runWorker   this.firstTask = firstTask;   this.thread = getThreadFactory().newThread(this); }

Worker就像是任务的一个代理,Worker实现了Runnable,所以要重写run方法,这个run方法调用了runWorker()

public void run() {   runWorker(this); }

Worker继承了AQS,那么它本身就是一个锁,在执行任务的时候,它会锁住它自己,执行完任务然后再释放自己,Worker里面也是重写了tryAcquire()tryRelease()

创建线程池及一些线上问题

第一种线程池的创建方式是利用线程池的工具类Executors来new一个这个线程池

ExecutorService threadPool = Executors.newFixedThreadPool(3) //3就是corePoolSize,核心线程数量

第二线程池的创建方式是直接new一个ThreadPollExecutor(),下面这种方式更推荐

Java源码

这边有几个参数

  • corePollSize:指定核心线程数量,这个就是线程池里
  • maximumPoolSize:这个是假设我线程池里已经有corePollSize个线程了,但是它们处理任务很慢,然后阻塞队列也满了,如果此时maximumPoolSize它是大于这个核心线程数的,那就会再创建这个线程出来执行任务,那么这个maximumPoolSize创建出来的线程什么时候被回收呢,就用到了下面这个参数
  • keepAliveTime:这个也就是说,我maximumPoolSize里面的线程空闲时间达到keepAliveTime,那就可以回收了
  • 还有一种情况是这个corePollSize满了,队列满了,maximumPoolSize也满了,那就不能再创建额外的线程出来了,所以这边就有拒绝策略,常见的拒绝策略是 AbortPolicy,就直接丢弃任务,还有DiscardPolicyDiscardOldestPolicy等等
  • 一般用的是fixed线程池,它的corePollSizemaximumPoolSize是一样大的,然后默认的阻塞队列是无界阻塞队列,可以无限制放任务,如果用无界阻塞队列不设置上限,然后处理任务也很慢,队列积压很多,那么内存飙升,回收也没法回收,就可能导致OOM
  • 针对上面的解决策略是,我可以自定义一个拒绝策略,队列满了,来任务了,我就持久化到磁盘上去,等我任务处理的差不多了,再从磁盘上读进来再处理任务。
  • 当然线程机器可能宕机,那么这个任务不就没了吗,所以一种处理措施是,我在提交任务的时候,往数据库里写上这个任务的信息,然后给他设置个状态,是未提交、提交了、已完成,提交成功后更改状态,如果此时系统宕机也没关系,系统重启再去扫描数据库里面未提交和已提交的任务,再执行

submit()和execute()

首先看一下结构

Executor是线程池最顶级的接口,它的一个子接口是ExecutorService,然后是有一个抽象类AbstractExecutorService实现了ExecutorService这个接口,最后才是这个ThreadPoolExecutor来继承这个抽象类。

public interface ExecutorService(二级接口) extends Executor(线程顶级接口) abstract class AbstractExecutorService(第三级) implements ExecutorService class ThreadPoolExecutor(第四级) extends AbstractExecutorService

但是我们new一个ThreadPoolExecutor出来后,既可以执行submit(),也可以执行execute(),实际上这两个是有区别的

  • 顶级接口Executor这个接口它只有一个execute()方法

    • public interface Executor {     void execute(Runnable command); }
  • 二级接口ExecutorService,在这一层,添加了submit()方法

    • command+F12

      Java源码

  • 到了第三层AbstractExecutorService,它实现了ExecutorService这个接口,自然有submitexecute这两个方法

  • 到了第四层ThreadPoolExecutor这个实现类,它没有submit这个方法,但是有execute方法

    • 那么实际上ThreadPoolExecutor的实例调用的是AbstractExecutorServicesubmit方法

    • //AbstractExecutorService的submit方法,举例两个 public <T> Future<T> submit(Callable<T> task) {     if (task == null) throw new NullPointerException();     RunnableFuture<T> ftask = newTaskFor(task);     execute(ftask);     return ftask; } public <T> Future<T> submit(Runnable task, T result) {     if (task == null) throw new NullPointerException();     RunnableFuture<T> ftask = newTaskFor(task, result);     execute(ftask);     return ftask; }
    • 可以看到这个AbstractExecutorService的submit方法本质上还是调用了Executor的execute方法

注意submit的返回值是 Future 类的,而execute的返回值是void

submit做了两件事情

  • submit的参数可以是实现了runnable接口的,也可以是实现了callable接口的,第一件事情就是把Runnable接口和Callable都转化成FutureTask
  • 第二件事情就是让execute方法执行FutureTask

execute()

那么ThreadPoolExecutor的execute()这个方法的实现是怎样的呢?

Java源码

这一道这边有个,这个意思是说,并发的时候,可能线程还没启动,任务已经跑起来了,所以会预先搞个非核心线程出来

else if (workerCountOf(recheck) == 0)       addWorker(null, false);

addWorker()与runWorker()

接着看一下addWorker()方法,它就是结合线程池的情况看要不要添加Worker

Java源码

注意addWorker方法传进来的参数是firstTask, 在初始化的过程中,会把 worker自己作为任务传给 thread 去初始化,上文说了 Worker 在初始化时的关键代码,this.thread = getThreadFactory ().newThread (this)Worker(this) 是作为新建线程的构造器入参的,所以 t.start () 会执行到 Worker 的 run 方法上

Java源码

这个worker的run调用了一个runWorker

Java源码

这个runWorker才是真正去执行run方法的,也是比较重要的方法

Java源码

通过w.firstTask拿到的task,此时的task是FutureTask类(Future接口的子接口又个RunnableFuture,这个RunnableFuture的父接口是Future和Runnable,然后FutureTask实现了RunnableFuture这个接口,所以FutureTask里面也有run方法),直接看FutureTask的run的源码,它是没有返回值的

Java源码

执行完任务之后呢

上面代码也看到了,在runWorker()方法中是一个while循环,如果队列中有任务就继续执行

Java源码

那么关键就是这个getTask方法

Java源码


Java集合源码

最短路算法这篇博客主要是记一下最短路板子,板子三天不练就忘(x 回到正题,图的最短路算法有很多,在此记录一下非常常用的三个算法 单源最短路 不带负权边:$Dijkstra$ 带负权边:$Bellman-Ford$、$SPFA$ 多源最
2020-03-10 算法
树状数组一、树状数组概念Binary Indexed Tree,用于维护前缀信息的结构,对前缀信息处理也十分高校,用于解决前缀信息问题和区间类问题。 比如:给定一个数组,实现两个函数 update(int index,int val),将
2019-12-12 算法

本文地址https://www.b2bchain.cn/?p=6051

赞(0) 打赏
部分文章转自网络,侵权联系删除b2bchain区块链学习技术社区 » Java源码
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

b2b链

联系我们联系我们