Introduction to Android thread pool

This article mainly uses kotlin to discuss the usage of thread pool in Android development.

When we want to use threads, we can directly create sub threads and start them

Thread { Log.d("rfDev","rustfisher said: hello") }.start()

In this case, we can consider thread pool. Thread pool solves two problems: when a large number of asynchronous tasks need to be executed, it reduces the call overhead of each asynchronous task and improves performance. In addition, it can limit and manage sub threads. Each ThreadPoolExecutor maintains statistics, such as the number of tasks executed.

When there are a large number of asynchronous tasks, you can consider using thread pool.

ThreadPoolExecutor provides many parameters for developers to control. The designer of thread pool suggests that developers use the following factory methods. There are mainly five in Android

In fact, when we enter executors.new in Android studio, many prompt options will pop up.

Smart tips for executors.new

Get a cacheable thread pool object with executors.newcachedthreadpool and let it execute the task.

val tp: ExecutorService = Executors.newCachedThreadPool()
tp.submit { Log.d(TAG,"rustfisher: cached线程池执行任务 3") }

The cacheable thread pool creates new child threads when needed. When the original thread is available, the existing thread will be reused. This mechanism is suitable for executing multiple short-term asynchronous tasks. The tasks are small, but the number is large.

Calling the execute method first attempts to reuse existing available threads. If there is no thread currently, a new thread is created and added to the pool. Threads that are not used for more than 60 seconds are stopped and removed. Therefore, even if the thread pool is not used for a long time, it will not cause much overhead.

Using the newfixedthreadpool (int nthreads) example

val fixedTp: ExecutorService = Executors.newFixedThreadPool(4)
fixedTp.submit { Log.d(TAG,"rustfisher 定长线程池执行任务") }

In the static method, an int parameter nthreads is passed in, indicating the maximum number of threads. If all threads are currently busy, a new task is added. Then the task waits in the queue until a thread is available to process the task.

If a thread encounters an error and stops, a new thread will be created to fill the position to execute the task.

Threads in the pool will survive until the thread pool stops (executorservice#shutdown).

val singleTp: ExecutorService = Executors.newSingleThreadExecutor()
singleTp.submit { Log.d(TAG,"单一线程池执行任务") }

Has only 1 child thread. Task queues do not limit the number of tasks. If the thread encounters a problem and stops, a new thread will be created to process the next task.

It can ensure that tasks will be processed in order, and only one task can be processed at the same time.

After a single thread pool is created, the number of threads cannot be dynamically modified. Unlike the fixed length thread pool of newfixedthreadpool (1), the number of threads can be modified.

val scheduleTp: scheduledexecutorservice = Executors.newScheduledThreadPool(3)

The scheduled task thread pool can execute delayed tasks and periodic tasks.

Delay and time unit need to be set

scheduleTp.schedule({ Log.d(TAG,"计划任务1 runnable") },300,TimeUnit.MILLISECONDS)
scheduleTp.schedule(Callable { Log.d(TAG,"计划任务2 callable") },400,TimeUnit.MILLISECONDS)

It mainly involves two methods, scheduleatfixedrate and schedulewithfixeddelay.

Assuming that the task time is less than the cycle time, it is carried out according to the given cycle time. The two methods are consistent.

Assuming that the task execution time is greater than the cycle time, the two methods are a little different

Android SDK is greater than or equal to 24. There is a new thread pool, which is temporarily called "work stealing thread pool" or "flexible scheduling thread pool".

if (android.os.Build.VERSION.SDK_INT >= android.os.Build.VERSION_CODES.N) {
    Executors.newWorkStealingPool()
}

The thread pool maintains enough threads to support a given parallelism level, and multiple queues may be used to reduce contention. Parallelism corresponds to the maximum number of active threads or the maximum number of threads that can process tasks.

The actual number of threads may increase or decrease dynamically. The work stealing thread pool does not guarantee that tasks are processed in the order of submission.

When executing a task, you can pass in runnable and callable. Both runnable are used earlier.

Callable example

tp.submit(Callable { "OK" })

Callable and runnable are OK for tasks without return value.

val tp: ExecutorService = Executors.newCachedThreadPool()
tp.submit { Log.d(TAG,"rustfisher: cached线程池submit runnable") }
tp.execute { Log.d(TAG,"rustfisher: cached线程池execute runnable") }
tp.submit(Callable { Log.d(TAG,"rustfisher: cached线程池submit callable") })

tp.shutdown() // 最后记得用完后停掉线程池

Callable interface is required for tasks with return value.

A future object is returned when the submit method is called. The return value can be obtained through the get () method of future. Note that get () is blocked. After completing the task, you can get the return value.

val tp: ExecutorService = Executors.newCachedThreadPool()
val future = tp.submit(Callable {
    return@Callable "callable的返回值"
})
Log.d(TAG,"future get之前 isDone: ${future.isDone},isCancelled: ${future.isCancelled}")
val res = future.get()
Log.d(TAG,"future get之后 isDone: ${future.isDone},isCancelled: ${future.isCancelled}")
Log.d(TAG,"future get: $res")

Run log

future get之前 isDone: false,isCancelled: false
future get之后 isDone: true,isCancelled: false
future get: callable的返回值

For tasks in the list, you can use invokeall (collection <? Extensions callable < T > > tasks) to return a future list. For comparison, add a delay to one of the tasks.

Invokeall example

    val tp: ExecutorService = Executors.newFixedThreadPool(5)
    val callList = arrayListOf<Callable<String>>(
            Callable {
                Log.d(TAG,"task1 ${Thread.currentThread()}")
                return@Callable "rust"
            },Callable {
                Log.d(TAG,"task2 ${Thread.currentThread()}")
                Thread.sleep(1500) // 加上延时
                return@Callable "fisher"
            },"task3 ${Thread.currentThread()}")
                return@Callable "列表里面的任务"
            },)
    Log.d(TAG,"invokeAll 准备提交任务")
    val futureList = tp.invokeAll(callList)
    Log.d(TAG,"invokeAll 已提交任务")
    futureList.forEach { f ->
        Log.d(TAG,"任务列表执行结果 ${f.get()}") // 这里会阻塞 别在ui线程里get
    }

Run the log and you can see that after the task is submitted, the running results are obtained after a delay. Pay attention to the time before and after invokeall. Invokeall blocks the current thread. You must be careful when using it, and do not call it in UI thread.

    2021-09-11 14:40:07.062 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: invokeAll 准备提交任务
    2021-09-11 14:40:07.063 16914-19230/com.rustfisher.tutorial2020 D/rfDevTp: task1 Thread[pool-4-thread-1,5,main]
    2021-09-11 14:40:07.063 16914-19231/com.rustfisher.tutorial2020 D/rfDevTp: task2 Thread[pool-4-thread-2,main]
    2021-09-11 14:40:07.063 16914-19232/com.rustfisher.tutorial2020 D/rfDevTp: task3 Thread[pool-4-thread-3,main]
    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: invokeAll 已提交任务
    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: 任务列表执行结果 rust
    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: 任务列表执行结果 fisher
    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: 任务列表执行结果 列表里面的任务

Three tasks are submitted and executed in three different sub threads.

Invokeany (collection <? Extensions callable < T > > tasks) is also a callable collection. Then return the value of the task that ends first. Other unfinished tasks will be cancelled normally without exception.

Invokeany example

    val tp: ExecutorService = Executors.newCachedThreadPool()
    val callList = arrayListOf<Callable<String>>(
            Callable {
                Thread.sleep(1000) // 设计延时
                return@Callable "rust"
            },Callable {
                Thread.sleep(400)
                return@Callable "fisher"
            },Callable {
                Thread.sleep(2000)
                return@Callable "列表里面的任务"
            },"invokeAny 提交任务")
    val res = tp.invokeAny(callList)
    Log.d(TAG,"执行结果 $res")

    2021-09-11 14:04:55.253 14066-14066/com.rustfisher.tutorial2020 D/rfDevTp: invokeAny 提交任务
    2021-09-11 14:04:55.654 14066-14066/com.rustfisher.tutorial2020 D/rfDevTp: 执行结果 fisher

You can see from the log that the "Fisher" task is executed at last.

After use, remember to terminate the thread pool

/*ExecutorService*/ shutdown()
shutdownNow()

Shutdown () creates a stop command after the submitted task and does not accept new tasks. If the thread pool has stopped, calling this method will not take effect.

The shutdownnow () method attempts to stop all executing tasks and stop waiting tasks. And return the list of tasks waiting to be executed list < runnable >.