此文已由作者赵计刚授权网易云社区发布。
欢迎访问网易云社区,了解更多网易技术产品运营经验。
3、newFixedThreadPool(int corePoolSize)
源代码:
说明:execute()的源代码查看第十三章 ThreadPoolExecutor源码解析
4、newSingleThreadExecutor()
源代码:
/** * 1、创建一个线程数固定(corePoolSize==maximumPoolSize==1)的线程池 * 2、核心线程会一直运行 * 3、无界队列LinkedBlockingQueue * 注意:所有task都是串行执行的 */ public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
说明:execute()的源代码查看第十三章 ThreadPoolExecutor源码解析
5、newCachedThreadPool()
源代码:
/** * 1、创建一个线程池:当池中的线程都处于忙碌状态时,会立即新建一个线程来处理新来的任务 * 2、这种池将会在执行许多耗时短的异步任务的时候提高程序的性能。 * 3、6秒钟内没有使用的线程将会被中止,并且从线程池中移除,因此几乎不必担心耗费资源 * 4、队列:SynchronousQueue * 5、maximumPoolSize为Integer.MAX_VALUE */ public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
说明:execute()的源代码查看第十三章 ThreadPoolExecutor源码解析
6、newScheduledThreadPool(int corePoolSize)
源代码:
Executors:newScheduledThreadPool(int corePoolSize)
/** * 创建一个线程池:该线程池可以用于执行延时任务或者定时任务 */ public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); }
ScheduledThreadPoolExecutor:ScheduledThreadPoolExecutor(int corePoolSize)
说明:ScheduledThreadPoolExecutor是ThreadPoolExecutor的子类,其中调用的super构造器就是ThreadPoolExecutor的构造器。
ScheduledThreadPoolExecutor:execute(Runnable command)
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); schedule(command, 0, TimeUnit.NANOSECONDS); }
ScheduledThreadPoolExecutor:schedule(Runnable command, long delay, TimeUnit unit)
/** * 这个方法:其实就是将task封装一下,然后加入到DelayedWorkQueue中 * 1、DelayedWorkQueue其实就是一个DelayQueue * 2、当有新的task加入时,DelayQueue会将其加入内部的数组对象中,并对其进行排序,在这里,排序的规则就是执行的时间,执行时间越近的排在越前 * 3、线程池中的线程在执行task时,获取最近要执行的task,然后唤醒所有等待available条件的线程来执行该任务 */ public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { if (command == null || unit == null) throw new NullPointerException(); RunnableScheduledFuture<?> t = decorateTask(command, new ScheduledFutureTask<Void>(command, null, triggerTime(delay, unit))); delayedExecute(t); return t; }
注意:这里的注释就是整个ScheduledThreadPoolExecutor的执行机理。
下面说一下其中调用到的一些方法。
第一部分:封装ScheduledFutureTask任务
ScheduledThreadPoolExecutor:triggerTime(long delay, TimeUnit unit)
/** * 返回一个delayed action(延时任务)的触发时间 */ private long triggerTime(long delay, TimeUnit unit) { return triggerTime(unit.toNanos((delay < 0) ? 0 : delay)); } /** * Returns the trigger time of a delayed action. */ long triggerTime(long delay) { return now() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay)); }
说明:用于计算延时任务的触发时间。
注意:在上边的execute()方法中传递的delay是0,根据上边的代码,计算出触发时间就是now()。
ScheduledThreadPoolExecutor:内部类ScheduledFutureTask
private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> { private final long sequenceNumber;//用于打破FIFO关系的序列号 private long time;//任务执行的触发时间 /** * 一个用于重复执行的任务的时间段(单位:纳秒) * 0-->不重复执行的任务 * 正值:fixed-rate执行 * 负值:fixed-delay执行 */ private final long period; /** * 创建一个一次性的action并且指定触发时间 */ ScheduledFutureTask(Runnable r, V result, long ns) { super(r, result); this.time = ns; this.period = 0; this.sequenceNumber = sequencer.getAndIncrement(); }
说明:ScheduledFutureTask是FutureTask的子类,上边的构造器中的super(r, result)代码如下:
FutureTask:FutureTask(Runnable runnable, V result)
private final Sync sync;//控制FutureTask的同步器 public FutureTask(Runnable runnable, V result) { sync = new Sync(Executors.callable(runnable, result)); }
Executors:callable(Runnable task, T result)
public static <T> Callable<T> callable(Runnable task, T result) { if (task == null) throw new NullPointerException(); return new RunnableAdapter<T>(task, result); }
Executors:内部类RunnableAdapter
注意:这里才是task真正去运行的地方。-->task.run()
至此,ScheduledFutureTask任务封装完成。
第二部分:修饰任务
ScheduledThreadPoolExecutor:RunnableScheduledFuture
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) { return task; }
说明:这里其实就是直接返回了刚刚封装好的任务
第三部分:将延时任务加入阻塞队列
ScheduledThreadPoolExecutor:delayedExecute(Runnable command)
说明:这样之后,之前封装好的任务就加入了延时队列DelayQueue(阻塞队列的一个子类)
DelayQueue:add(E e)
public boolean add(E e) { return offer(e); } public boolean offer(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { E first = q.peek();//获取队列头部节点但不删除 q.offer(e);//将e放到q的尾部 //如果队列中只有e或者e的触发时间小于队头结点 if (first == null || e.compareTo(first) < 0) available.signalAll(); return true; } finally { lock.unlock(); } }
说明:在该方法中,将上边封装好的任务就加入了DelayQueue,并将该任务置于了队头,然后唤醒所有等待available条件的线程来执行该任务。
总结:
免费领取验证码、内容安全、短信发送、直播点播体验包及云服务器等套餐
更多网易技术、产品、运营经验分享请点击。