Android开发
Android架构师必备技能 | 并发编程之线程池
线程池
Java的线程池是运用场景最多的并发框架，几乎所有需要异步或者并发执行任务的程序都可以使用线程池。
使用线程池能带来的好处：
- 降低资源消耗
- 提高响应速度
- 提高线程的可管理性
线程池底层实现分析
Java 中提供了四种线程池创建方法，分别是：
-
newSingleThreadExecutor
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。 -
newFixedThreadPool
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。 -
newScheduledThreadPool
创建一个可定期或者延时执行任务的定长线程池，支持定时及周期性任务执行。 -
newCachedThreadPool
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
1，newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
可以看到 corePoolSize、maximumPoolSize 都是 1 ，keepAliveTime 是 0 ，使用的是 LinkedBlockingQueue 队列。
解释：创建只有一个线程的线程池，且线程的存活时间是无限的；当该线程正繁忙时，对于新任务会进入阻塞队列中(无界的阻塞队列)
demo：
public class FourThreadPool {
public static void main(String[] args) {
ScheduledThreadPoolExecutorTest().ScheduledThreadPoolExecutorTest();
ExecutorService executorService = new newSingleThreadExecutor().newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
executorService.execute(new TestRunnable());
}
executorService.shutdown();
}
}
class TestRunnable implements Runnable {
static int i = 1;
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 线程被调用了。第" + getCount() + "次");
}
public static int getCount() {
return i++;
}
}
class newSingleThreadExecutor {
ExecutorService newSingleThreadExecutor() {
/**
* 1
* 此线程池 Executor 只有一个线程。它用于以顺序方式的形式执行任务。
* 如果此线程在执行任务时因异常而挂掉，则会创建一个新线程来替换此线程，后续任务将在新线程中执行。
*/
ExecutorService executorService = Executors.newSingleThreadExecutor();
return executorService;
}
}
输出：
pool-1-thread-1 线程被调用了。第1次 pool-1-thread-1 线程被调用了。第2次 pool-1-thread-1 线程被调用了。第3次 pool-1-thread-1 线程被调用了。第4次 pool-1-thread-1 线程被调用了。第5次 pool-1-thread-1 线程被调用了。第6次 pool-1-thread-1 线程被调用了。第7次 pool-1-thread-1 线程被调用了。第8次 pool-1-thread-1 线程被调用了。第9次 pool-1-thread-1 线程被调用了。第10次
2，newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
可以看到 corePoolSize 和 maximumPoolSize 一样， 由调用者自己设定传入 ，keepAliveTime 是 0 ，使用的是 LinkedBlockingQueue 队列。
解释：创建可容纳固定数量线程的池子，每隔线程的存活时间是无限的，当池子满了就不在添加线程了；如果池中的所有线程均在繁忙状态，对于新任务会进入阻塞队列中(无界的阻塞队列)
可以看到上面两种线程池都是使用了 LinkedBlockingQueue ，而且默认使用的是 new LinkedBlockingQueue<Runnable>());
继续扒一下这个 LinkedBlockingQueue 的默认大小
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
竟然是 int 类型的最大值
@Native public static final int MAX_VALUE = 0x7fffffff;
就是说这个队列里面可以放 2^31-1 = 2147483647 个 任务。
demo:
public class FourThreadPool {
public static void main(String[] args) {
ScheduledThreadPoolExecutorTest().ScheduledThreadPoolExecutorTest();
ExecutorService executorService = new newFixedThreadPool().newFixedThreadPool();
for (int i = 0; i < 10; i++) {
executorService.execute(new TestRunnable());
}
executorService.shutdown();
}
}
class TestRunnable implements Runnable {
static int i = 1;
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 线程被调用了。第" + getCount() + "次");
}
public static int getCount() {
return i++;
}
}
class newFixedThreadPool {
ExecutorService newFixedThreadPool() {
/**
* 2
* 它是一个拥有固定数量线程的线程池。提交给 Executor 的任务由固定的 n 个线程执行，
* 如果有更多的任务，它们存储在 LinkedBlockingQueue 里。这个数字 n 通常跟底层处理器CPU支持的线程总数有关。
*/
ExecutorService executorService = Executors.newFixedThreadPool(3);
return executorService;
}
}
输出：
pool-1-thread-1 线程被调用了。第1次 pool-1-thread-2 线程被调用了。第2次 pool-1-thread-2 线程被调用了。第4次 pool-1-thread-2 线程被调用了。第5次 pool-1-thread-2 线程被调用了。第6次 pool-1-thread-2 线程被调用了。第7次 pool-1-thread-1 线程被调用了。第3次 pool-1-thread-1 线程被调用了。第10次 pool-1-thread-2 线程被调用了。第9次 pool-1-thread-3 线程被调用了。第8次
可以看到最多只有 3 个线程，哪个线程空闲了就执行，超过 corePoolSize=3后，任务会被丢到 LinkedBlockingQueue 队列，因为LinkedBlockingQueue 很大，所以执行完前面的任务，空闲后会执行队列 LinkedBlockingQueue 里面的任务。
3，newScheduledThreadPool
public class ScheduledThreadPoolExecutor
extends ThreadPoolExecutor
implements ScheduledExecutorService {
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
这里的 super 还是会调用 ThreadPoolExecutor 的方法。
可以看到 corePoolSize 由调用者自己设定传入 ，maximumPoolSize 是 int的最大值，keepAliveTime 是 0 ，使用的是 DelayedWorkQueue 队列。
new DelayedWorkQueue() : 一个按超时时间升序排序的队列，顾名思义，可以延迟执行。
我们可以自定义设置队列的延迟策略：
* @param command the task to execute
* @param initialDelay the time to delay first execution
* @param period the period between successive executions
* @param unit the time unit of the initialDelay and period parameters
* @return a ScheduledFuture representing pending completion of
* the task, and whose {@code get()} method will throw an
* exception upon cancellation
* @throws RejectedExecutionException if the task cannot be
* scheduled for execution
* @throws NullPointerException if command is null
* @throws IllegalArgumentException if period less than or equal to zero
*/
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
demo：
class ScheduledThreadPoolExecutorTest {
ScheduledExecutorService ScheduledThreadPoolExecutorTest() {
/**
* 3
* 当我们有一个需要定期运行的任务或者我们希望延迟某个任务时，就会使用此类型的 executor。
*/
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//创建并执行在给定延迟后启用的单次操作。延迟2秒后启动，但是只执行一次。
// executorService.schedule(new TestRunnable(), 2, TimeUnit.SECONDS);
//延迟2秒后启动，之后每间隔1秒执行一次
executorService.scheduleAtFixedRate(new TestRunnable(), 2, 1, TimeUnit.SECONDS);
return executorService;
}
public static void main(String[] args) {
new ScheduledThreadPoolExecutorTest().ScheduledThreadPoolExecutorTest();
}
}
class TestRunnable implements Runnable {
static int i = 1;
@Override
public void run() {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
System.out.println(Thread.currentThread().getName() + " 线程被调用了。第" + getCount() + "次");
}
public static int getCount() {
return i++;
}
}
输出：
2020-07-15 18:03:47 2020-07-15 18:03:49 pool-1-thread-1 线程被调用了。第1次 2020-07-15 18:03:50 pool-1-thread-1 线程被调用了。第2次 2020-07-15 18:03:51 pool-1-thread-2 线程被调用了。第3次 2020-07-15 18:03:52 pool-1-thread-2 线程被调用了。第4次 2020-07-15 18:03:53 pool-1-thread-2 线程被调用了。第5次 2020-07-15 18:03:54 pool-1-thread-2 线程被调用了。第6次 2020-07-15 18:03:55 pool-1-thread-2 线程被调用了。第7次
4，newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
corePoolSize为0；maximumPoolSize 为 Integer.MAX_VALUE；keepAliveTime 为 60L ，单位是秒，使用 SynchronousQueue 队列。
解释：SynchronousQueue 是同步队列，因此会在池中寻找可用线程来执行，若有可以线程则执行，若没有可用线程则创建一个线程来执行该任务；若池中线程空闲时间超过指定大小，则该线程会被销毁。
适用：执行很多短期异步的小程序或者负载较轻的服务器
demo：
public class FourThreadPool {
public static void main(String[] args) {
ScheduledThreadPoolExecutorTest().ScheduledThreadPoolExecutorTest();
ExecutorService executorService = new newCachedThreadPoolTest().newCachedThreadPoolTest();
for (int i = 0; i < 10; i++) {
executorService.execute(new TestRunnable());
}
executorService.shutdown();
}
}
class TestRunnable implements Runnable {
static int i = 1;
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 线程被调用了。第" + getCount() + "次");
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static int getCount() {
return i++;
}
}
class newCachedThreadPoolTest {
ExecutorService newCachedThreadPoolTest() {
/**
* 4
* 此线程池的线程数不受限制。如果所有的线程都在忙于执行任务并且又有新的任务到来了，这个线程池将创建一个新的线程并将其提交到 Executor。
* 只要其中一个线程变为空闲，它就会执行新的任务。 如果一个线程有 60 秒的时间都是空闲的，它们将被结束生命周期并从缓存中删除。
*/
ExecutorService executorService = Executors.newCachedThreadPool();
return executorService;
}
}
执行结果：
pool-1-thread-2 线程被调用了。第1次 pool-1-thread-1 线程被调用了。第2次 pool-1-thread-3 线程被调用了。第3次 pool-1-thread-4 线程被调用了。第4次 pool-1-thread-5 线程被调用了。第5次 pool-1-thread-6 线程被调用了。第6次 pool-1-thread-7 线程被调用了。第7次 pool-1-thread-8 线程被调用了。第8次 pool-1-thread-9 线程被调用了。第9次 pool-1-thread-10 线程被调用了。第10次
线程池任务执行流程 ：
-
当线程池小于 corePoolSize 时，新提交任务将创建一个新线程执行任务，即使此时线程池中存在空闲线程；
-
当线程池达到 corePoolSize 时，新提交任务将被放入 workQueue 中，等待线程池中任务调度执行；
-
当 workQueue 已满，且
maximumPoolSize>corePoolSize时，新提交任务会创建新线程执行任务； -
当提交任务数超过 maximumPoolSize 时，新提交任务由RejectedExecutionHandler 处理；
-
当线程池中超过 corePoolSize 线程，空闲时间达到 keepAliveTime 时，关闭空闲线程；
-
当设置
allowCoreThreadTimeOut(true)时，线程池中 corePoolSize 线程空闲时间达到 keepAliveTime 也将关闭；
工作队列排队策略
1、已经说过当线程池中工作线程的总数量超过核心线程数量后，新加的任务就会放入工作队列中进行等待被执行
2、使用线程池就得创建ThreadPoolExecutor对象，通过ThreadPoolExecutor(线程池)类的构造方法创建时，就得指定工作队列，它是BlockingQueue
SynchronousQueue 直接提交
-
直接提交策略----意思是工作队列不保存任何任务被等待执行，而是直接提交给线程进行执行。
-
工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保存它们。
-
如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。
-
此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。
-
Executors 的
newCacheThreadPool()方法创建线程池，就是使用的此种排队策略
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
LinkedBlockingQueue 无界队列
-
无界队列策略----无界指的是工作队列大小没有上限，可以添加无数个任务进行等待。
-
使用无界队列将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。于是创建的线程就不会超过 corePoolSize。因此，maximumPoolSize 的值也就无效了。所以一般让corePoolSize等于maximumPoolSize
-
当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列
-
Executors 的
newFixedThreadPool(int nThreads)方法创建线程池，就是使用的此种排队策略
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
ArrayBlockingQueue 有界队列
- 有界队列策略----意思是工作队列的大小是有限制的
- 优点是可以防止资源耗尽的情况发生，因为如果工作队列被无休止的添加任务也是很危险的
- 当工作队列排满后，就会执行线程饱和策略
// 构造线程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 4,
3, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2),
new ThreadPoolExecutor.DiscardOldestPolicy());
-
如上核心线程为3个，每个线程的工作队列大小为2(即队列中最多有两个任务在等待执行)，线程池最大线程数为4个
-
所以当工作线程数小于等于3时，直接新建线程执行任务；超过3时，任务会被添加进工作队列进行等待，3*2=6，当工作队列等待的任务数超过6个以后，则又会新建一个线程，此时整个线程池线程总数已经达到了4个，当还有任务进行添加时，此时将采取饱和策略。
手写简单线程池
实现一个简单的线程池的demo案例：
package main;
import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class MyThreadPool2 {
// 线程池中线程个数为5
private static int WORK_NUM = 5;
// 队列中的队列任务为100
private static int TASK_COUNT = 100;
// 工作线程组
private WorkThread[] workThreads;
// 任务队列
private BlockingQueue<Runnable> taskQueue = null;
private final int worker_num;// 用户构造这个线程池的时候希望启用的线程组
// 根据个数创建默认的线程池
public MyThreadPool2() {
this(WORK_NUM, TASK_COUNT);
}
@Override
public String toString() {
return "MyThreadPool2 [workThreads=" + Arrays.toString(workThreads) + ", taskQueue=" + taskQueue
+ ", worker_num=" + worker_num + "]";
}
public MyThreadPool2(int worker_num, int taskCount) {
if (WORK_NUM <= 0) {
worker_num = WORK_NUM;
}
if (taskCount <= 0) {
taskCount = TASK_COUNT;
}
this.worker_num = worker_num;
taskQueue = new ArrayBlockingQueue<>(taskCount);
workThreads = new WorkThread[worker_num];
for (int i = 0; i < worker_num; i++) {
workThreads[i]=new WorkThread();
workThreads[i].stopWorker();
workThreads[i].start();
}
}
// 执行任务，把任务加入队列中，什么时候执行由线程池来决定
public void execute(Runnable task) {
try {
taskQueue.put(task);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public void destory()
{
System.out.println("ready close pool");
for (int i=0;i<worker_num;i++)
{
workThreads[i].stopWorker();
workThreads[i]=null;
}
taskQueue.clear();
}
/**
* 内部类，工作线程
*
* @author xuyuanfeng
*
*/
private class WorkThread extends Thread {
public WorkThread() {
// TODO Auto-generated constructor stub
}
@Override
public void run() {
Runnable r = null;
// 当前线程有没有被终止
while (!isInterrupted()) {
try {
r = taskQueue.take();
if (r != null) {
System.out.println(getId() + " ready exec:" + r);
r.run();
}
r = null;
} catch (InterruptedException e) {
e.printStackTrace();
} // 拿到任务
}
}
//中断线程
public void stopWorker() {
interrupt();
}
}
}
package main;
import java.util.Random;
public class Main {
public static void main(String[] args) throws InterruptedException {
MyThreadPool2 t = new MyThreadPool2(3, 0);
t.execute(new MyThread("testA"));
t.execute(new MyThread("testB"));
t.execute(new MyThread("testC"));
t.execute(new MyThread("testD"));
t.execute(new MyThread("testE"));
Thread.sleep(10000);
t.destory();
System.out.println(t.toString());
}
static class MyThread implements Runnable {
private String name;
private Random r = new Random();
public MyThread(String name) {
this.name = name;
}
public String getName() {
return name;
}
@Override
public void run() {
try {
Thread.sleep(r.nextInt(1000) + 2000);
} catch (InterruptedException e) {
System.out.println("线程断开了");
// e.printStackTrace();
}
System.out.println("任务" + name + "完成");
}
}
}
测试结果
参考：https://www.cnblogs.com/zincredible/p/10984459.html https://blog.csdn.net/wangmx1993328/article/details/80582803 https://blog.csdn.net/RAVEEE/article/details/91353665
-
android.permission.read_media_video01-18
-
android_getaddrinfo failed eai_nodata01-18
-
androidmo01-18
-
Android下三种离屏渲染技术01-15
-
Android 蓝牙使用01-09
-
Android对接华为AI - 文本识别01-06
-
代码安全之代码混淆及加固（Android）11-15
-
简述Android语音播报TTS11-10
-
Android WiFi工具类11-06
-
Android开发未来的出路07-22
-
Android JNI 学习实践06-28
-
Android13 安装最新版 Frida06-26
-
Android Studio中SQLite的使用，主要介绍sqlite插入和读出图片（ViewBinder）的操作方法06-20
-
Android-NDK开发——基本概念06-18
-
Android Studio历史版本下载地址汇总06-04
