标签:abort hash shu 签名 BMI arraylist JD ecc 容器
在使用线程池后,创建线程变成了从线程池里获得空闲线程,关闭线程变成了将线程归坏给线程池。
JDK有一套Executor框架,大概包括Executor、ExecutorService、AbstractExeccutorService、ThreadPoolExecutor、Executors等成员,位于java.util.concurrent
包下。它们之间的关系如下:
Executor是顶层的接口,ExecutorService接口继承了它,AbstrctExecutorService继承了ExecutorService,ThreadPoolExecutor继承了AbstrctExecutorService。如果用<——
表示继承,<--
表示实现接口,它们的关系可表示如下:
Executor(接口) <—— ExecutorService(接口) <-- AbstrctExecutorService(抽象类) <—— ThreadPoolExecutor(类)
Executors是单独的一个类,可以看成是“线程池工厂”,它有很多静态方法,比如:
newFixedThreadPool该方法返回一个固定线程数的线程池。当有新任务提交时,如果线程池中有空闲线程就立即执行,否则会进入任务队列中,等到有空闲线程了才能执行。
newSingleThreadExecutor,该方法返回只有一个线程的线程池,处理策略和上面一样。实际上就是上面的参数指定为1而已。
newCachedThreadPool该方法返回一个可根据实际情况调整线程数的线程池,任务提交后,如果有空闲线程可以复用,则优先复用。若线程池中的线程全部在工作,而此时有新任务,则会创建新的线程来处理任务,所有线程执行完后会将线程归还给线程池。
newScheduledThreadPool返回一个ScheduledExecutorService对象,可以有计划地执行任务,比如在某个延时之后开始执行,或者周期性地执行某个任务。可以指定线程数量。
newSingleThreadScheduledExecutor实现了和上面一样的功能,不过线程池的大小为1。
ScheduledExecutorService有三个方法可以有计划地执行任务。如:
schedule(Runnable command, long delay, TimeUnit unit);
该方法可以在给定的延时后,执行一个任务;scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);
该方法以任务开始执行的时间为initialDelay,加上周期period,就是下一个任务开始执行的时间,以此类推,因此这个方法任务调度的频率是一定的;scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);
该方法表示每执行完一个任务,延迟delay
的时间后,开始执行下一个任务,initialDelay
还是表示任务开始的初始时延,上一个任务结束的时间点与下一个任务开始的时间点之差是固定的,固定为delay。即使单个任务的执行时间超过调度周期,scheduleAtFixedRate也不会让多个任务堆叠,比如任务执行需要8s,而调度周期是2s,调度第二个任务时,第一个还没执行完,因此为了避免任务堆叠,此时调度周期会变成8s;而采用scheduleWithFixedDelay,两个任务之间的实际间隔会变成10s,8s的执行+2s的delay。
这三个内部都是通过返回ThreadPoolExecutor产生线程池的。所以我们来重点关注它的构造方法。
public ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
workQueue是一个BlockingQueue接口的对象,存放的是Runnable对象。根据功能的不同,ThreadPoolExecutor中可以使用以下几种BlockingQueue
newFixedThreadPool因为它的corePoolSize和maxPoolSize大小一样,固定大小的线程不存在当实际线程数超过corePoolSize时要新增线程的可能,所以它使用了LinkedBlockingQueue,当有新任务且实际线程数已经达到最大时,会直接进入等待队列。
newSingleThreadExecutor是newFixedThreadPool的一种特殊情况,即取corePoolSize和maxPoolSize都为1
而newCachedThreadPool的corePoolSize为0,maxPoolSize为Integer.MAX_VALUE
,任务队列使用SynchronousQueue直接提交,新任务提交后,若有空闲线程就直接用,若没有就进入等待队列——但是这是个直接提交的队列,所有会新增线程执行该任务!由于corePoolSize为0,所以任务执行完毕后60s(构造函数指定)就会被回收。
当实际线程数超过maxPoolSize时,该采取什么样的策略?
下面看看CallerRunPolicy怎么拒绝的
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
DiscardOldestPolicy是这样做的
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll(); // 最老的一个请求在队列头部
e.execute(r);
}
}
ThreadFactory只有一个方法Thread newThread(Runnable r);
,线程池中的线程就是由它创建的。
fork也就是分支、分叉的意思,可以将大任务分解成小任务;join表示等待的意思,必须等待fork后的小任务执行完毕,得到执行后的部分结果,才能将部分结果合并成最终结果。
比如计算1到10000的和,就可以分成10个分支,每个分支计算一千个数的和,得到一个部分和,等待这10个部分和的结果都计算完毕,最后将其全部合并,得到最终的结果。
通常一个物理线程需要处理多个逻辑任务,所以每一个线程都有一个任务队列。若线程A的任务都执行完了,B还有很多任务没执行,此时A会“帮助”B执行它的任务,A帮助B执行B的任务时,从队列的尾部拿数据;而B自己执行任务时从队列头部拿数据,这就像是两个指针一个往左移动一个往右移动,避免了A、B之间对数据的竞争。
JDK中有ForkJoinPool,该接口有个方法public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task)
ForkJoinTask支持fork()
和join()
方法,它有两个重要的子类,没有返回值的RecursiveAction和有返回值的RecursiveTask,它们都有个方法compute()
,在这个方法中进行主要的计算。对于RecursiveAction来说签名是void,而对于RecursiveTask来说有返回值所以签名是<T>
CopyOnWriteArrayList的原理主要是:读的时候正常读,写-写需要同步,所以在写之前要使用Lock,然后为了读-写不阻塞,CopyOnWriteArrayList在写入操作时,先将原数组复制一份,然后在新数组末尾追加要添加的值,添加成功后再用新数组覆盖旧数组。
JDK中的该类的add方法是这样实现的:
public boolean add(E e) {
final ReentrantLock lock = this.lock;
// 保证写-写阻塞,故进行同步
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
// 关键!写入之前先赋值一个副本
Object[] newElements = Arrays.copyOf(elements, len + 1);
// 新数组的末尾添加
newElements[len] = e;
// 新数组覆盖旧数组
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
而数组的定义是这样的:
private transient volatile Object[] array;
注意有volatile关键字,说明当写数据的线程修改数组后,其他读取线程能立即“察觉”到。
BlockingQueue可以在并发环境下高效传输数据,本质上还是一个队列,数据从队列尾部入,从队列头部出。队列都有的offer()
和pull()
就不说了,没什么特别的。BlockingQueue还有put()
和take()
方法,正是这两个方法实现了阻塞。
以ArrayBlockingQueue来说:当队列为空时,take()方法会等待,直到队列不为空;当队列满时,put()方法会等待,直到队列有空闲位置。这是怎么实现的呢?来看代码
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
首先读和写都用的同一个锁lock,因此任何时候读和写只能有一个在执行。然后是条件notNull,等待非满,以便put;notEmpty等待非空,以便take。
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 关键,若队列满了,就等待
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
// 关键!一旦插入了数据,队列就不是非空了,于是唤醒在notEmpty上等待的线程(通知其他线程可以进行take啦)
notEmpty.signal();
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 关键!若队列为空,等待
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
// 关键!有元素出列了,等待在notFull上的线程可以被唤醒,可以进行put操作了
notFull.signal();
return x;
}
LinkedBlockingQueue和ArrayBlockingQueue原理大同小异,不过LinkedBlockingQueue读和写分别用一把锁,因此读和写可以同时进行。
/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();
ConcurrentSkipMap使用跳表实现。是一种可以进行快速查找的数据结构,时间复杂度是$O(lg n)$
跳表形象点说像个“直角三角形一样的金字塔”,每一层都是一条链表,最底层的链表包含了Map中的所有数据,每上一层都是下面一层的子集,越到上面结点越少。层与层之间通过值相同的元素链接起来,因此结点除了有指向本层的下一个结点的right,还有指向下层中具有相同值的元素的down(实际上通过数据结构Index表示)。另外,跳表中所有链表的元素都是排序的。
查找时,先从顶层开始查找,如果找到就结束了;否则当发现查找的值大于当前层的最大值(链表末尾),就会“跳到”下一层链表接着向前查找,查找朝着下面和右面两个方向进行,有点像“下台阶”...
by @sunhaiyu
2108.4.26
标签:abort hash shu 签名 BMI arraylist JD ecc 容器
原文地址:https://www.cnblogs.com/sun-haiyu/p/8953713.html