问题
在应用开发的多线程问题中，可能会遇到AsyncTask.execute()时会出现RejectedExecution的场景
java.util.concurrent.RejectedExecutionException: Task android.os.AsyncTask$SerialExecutor$1@7840c01 rejected from java.util.concurrent.ThreadPoolExecutor@5a73928[Running, pool size = 9, active threads = 9, queued tasks = 128, completed tasks = 192]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2014)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:794)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1340)
at android.os.AsyncTask$SerialExecutor.scheduleNext(AsyncTask.java:247)
at android.os.AsyncTask$SerialExecutor.execute(AsyncTask.java:241)
at android.os.AsyncTask.executeOnExecutor(AsyncTask.java:607)
at android.os.AsyncTask.execute(AsyncTask.java:551)
要从根源上解决以上问题，需要排查抛出异常时队列中任务的具体情况。
如何解决
整体设计如下：
image
- sDefaultExecutor在原生代码里就是SerialExecutor，而SerialExecutor是串行执行队列，但最终也是通过THREAD_POOL_EXECUTOR完成。为了加大复现概率，可以在Application创建时通过反射，替换AsyncTask的sDefaultExecutor为THREAD_POOL_EXECUTOR
if (BuildConfig.DEBUG) {
AsyncTaskWithTag.setEnableTaskInfo(true);
// 线程池监控
ReflectUtils.setStaticFieldValue(AsyncTask.class, "THREAD_POOL_EXECUTOR", AsyncTaskWithTag.THREAD_POOL_EXECUTOR);
ReflectUtils.setStaticFieldValue(AsyncTask.class, "sDefaultExecutor", AsyncTaskWithTag.THREAD_POOL_EXECUTOR);
}
- 使用ThreadPoolExecutorWithTag替换AsyncTask的sDefaultExecutor
//AsyncTaskWithTag.java
private static final ThreadFactory sThreadFactory = new ThreadFactory() {
private final AtomicInteger mCount = new AtomicInteger(1);
public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable, "AsyncTaskTag #" + mCount.getAndIncrement());
AsyncTaskRunningInfo.addTaskNameToMap(thread.toString(), runnable.toString());
return thread;
}
};
private static final RejectedExecutionHandler defaultHandler =
new RejectedExecutionHandlerWithTag();
public static final Executor THREAD_POOL_EXECUTOR;
static {
ThreadPoolExecutorWithTag threadPoolExecutor = new ThreadPoolExecutorWithTag(
CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
sPoolWorkQueue, sThreadFactory, defaultHandler);
threadPoolExecutor.allowCoreThreadTimeOut(true);
THREAD_POOL_EXECUTOR = threadPoolExecutor;
}
其中所有通过ThreadPoolExecutorWithTag创建的Thread，都将信息保存到AsyncTaskRunningInfo的Map集合中。
OOM时拒绝策略，使用自定义的RejectedExecutionHandlerWithTag。
public class RejectedExecutionHandlerWithTag implements RejectedExecutionHandler {
private static final String TAG = "Rejected";
private static final String TASK_CREATE_BY = " create by: ";
private static final String TASK_CREATE_TIME = " create at: ";
private static final String TASK_COST_TIME = " cost time: ";
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if (!executor.isShutdown()) {
prepareThreadInfo(executor);
prepareWorksQueueInfo(executor);
}
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
executor.toString());
}
// 工作线程
private void prepareThreadInfo(ThreadPoolExecutor executor) {
Object workersValue = ReflectUtils.getFieldValue(executor, "workers");
if (workersValue instanceof HashSet) {
long currentTime = System.currentTimeMillis();
StringBuilder builder = new StringBuilder();
HashSet<Object> workers = (HashSet<Object>) workersValue;
for (Object ob : workers) {
try {
Object threadValue = ReflectUtils.getFieldValue(ob, "thread");
if (threadValue instanceof Thread) {
Thread thread = (Thread) threadValue;
String name = thread.toString();
String taskName = AsyncTaskRunningInfo.getTaskNameFromMap(name);
if (taskName != null) {
// thread 对应的 task
String taskTag = AsyncTaskRunningInfo.getTaskNameFromMap(taskName);
if (taskTag != null) {
builder.append(name).append(TASK_CREATE_BY).append(taskTag);
} else {
taskName = name;
builder.append(name).append(TASK_CREATE_BY).append(taskName);
}
long taskTime = AsyncTaskRunningInfo.getTaskTimeFromMap(taskName);
if (taskTime > 0) {
builder.append(TASK_CREATE_TIME).append(taskTime);
builder.append(TASK_COST_TIME).append(currentTime - taskTime);
}
} else {
builder.append(name);
}
builder.append("\n");
}
} catch (Exception e) {
e.printStackTrace();
}
}
Log.d(TAG, " executor running thread queue : \n" + builder.toString());
}
}
// 等待队列
private void prepareWorksQueueInfo(ThreadPoolExecutor executor) {
long currentTime = System.currentTimeMillis();
StringBuilder builder = new StringBuilder();
for (Runnable runnable : executor.getQueue()) {
String name = runnable.toString();
String taskName = AsyncTaskRunningInfo.getTaskNameFromMap(name);
if (taskName != null) {
builder.append(name).append(TASK_CREATE_BY).append(taskName);
} else {
builder.append(name);
}
long taskTime = AsyncTaskRunningInfo.getTaskTimeFromMap(name);
if (taskTime > 0) {
builder.append(TASK_CREATE_TIME).append(taskTime);
builder.append(TASK_COST_TIME).append(currentTime - taskTime);
}
builder.append("\n");
}
Log(TAG, " executor pool work wait queue : \n" + builder.toString());
}
private void Log(String tag, String msg) {
if (tag == null || tag.length() == 0
|| msg == null || msg.length() == 0)
return;
// 每段打印3K
int segmentSize = 3 * 1024;
long length = msg.length();
if (length > segmentSize) {
while (msg.length() > segmentSize) {
String logContent = msg.substring(0, segmentSize);
Log.d(tag, logContent);
msg = msg.substring(segmentSize);
}
}
Log.d(tag, msg);
}
}
当触发OOM时回调到rejectedExecution,输出当前Map中的所有Task信息。
public class ThreadPoolExecutorWithTag extends ThreadPoolExecutor {
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
}
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}
@Override
protected void beforeExecute(Thread wt, Runnable task) {
super.beforeExecute(wt, task);
if (wt != null) {
AsyncTaskRunningInfo.removeTaskNameFromMap(wt.toString());
AsyncTaskRunningInfo.removeTaskTimeFromMap(wt.toString());
}
if (task != null) {
AsyncTaskRunningInfo.removeTaskNameFromMap(task.toString());
AsyncTaskRunningInfo.removeTaskTimeFromMap(task.toString());
}
}
@Override
public void execute(Runnable command) {
super.execute(command);
if (command == null)
throw new NullPointerException();
if (command instanceof AsyncTaskWithTag) {
AsyncTaskRunningInfo.addTaskNameToMap(command.toString(), ((AsyncTaskWithTag<?, ?, ?>) command).tagName());
}
AsyncTaskRunningInfo.addTaskTimeToMap(command.toString(), System.currentTimeMillis());
}
}
- AsyncTaskWithTag是继承于AsyncTask，使用AsyncTaskWithTag创建任务必须指定TagName，创建任务后会将runnable和TagName关联，在出现RejectedExecution时可通过runnable找到对应Task
public abstract class AsyncTaskWithTag<Params, Progress, Result> extends AsyncTask<Params, Progress, Result> {
private static final String TAG = "AsyncTaskWithTag";
private static boolean ENABLE_TASK_INFO = false;
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
// We want at least 2 threads and at most 4 threads in the core pool,
// preferring to have 1 less than the CPU count to avoid saturating
// the CPU with background work
private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
private static final int KEEP_ALIVE_SECONDS = 30;
private static final int QUEUE_SIZE = 128;
private static final BlockingQueue<Runnable> sPoolWorkQueue =
new LinkedBlockingQueue<>(QUEUE_SIZE);
public static final Executor THREAD_POOL_EXECUTOR;
private static final ThreadFactory sThreadFactory = new ThreadFactory() {
private final AtomicInteger mCount = new AtomicInteger(1);
public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable, "AsyncTaskTag #" + mCount.getAndIncrement());
AsyncTaskRunningInfo.addTaskNameToMap(thread.toString(), runnable.toString());
return thread;
}
};
private static final RejectedExecutionHandler defaultHandler =
new RejectedExecutionHandlerWithTag();
static {
ThreadPoolExecutorWithTag threadPoolExecutor = new ThreadPoolExecutorWithTag(
CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
sPoolWorkQueue, sThreadFactory, defaultHandler);
threadPoolExecutor.allowCoreThreadTimeOut(true);
THREAD_POOL_EXECUTOR = threadPoolExecutor;
}
public abstract String tagName();
public static boolean isEnableTaskInfo() {
return ENABLE_TASK_INFO;
}
public static void setEnableTaskInfo(boolean enableTaskInfo) {
ENABLE_TASK_INFO = enableTaskInfo;
}
public AsyncTaskWithTag() {
super();
if (isEnableTaskInfo()) {
prepareTaskInfo();
}
}
private void prepareTaskInfo() {
try {
Object future = ReflectUtils.getFieldValue(this, "mFuture");
if (future != null) {
AsyncTaskRunningInfo.addTaskNameToMap(future.toString(), tagName());
Log.d(TAG, "create task from " + tagName() + ":" + future.toString());
} else {
Log.d(TAG, "do not find future --- " + tagName());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
- 项目工程有较多使用sDefaultExecutor、THREAD_POOL_EXECUTOR的AsyncTask，可以考虑把这类AsyncTask全部替换为AsyncTaskWithTag。也可根据项目情况，先把某些嫌疑较大的AsyncTask替换为AsyncTaskWithTag，一步一步排除AsyncTask
new AsyncTask<Void, Void, Void>() {
@Override
protected Void doInBackground(Void... voids) {
...
}
}.execute();
修改为：
new AsyncTaskWithTag<Void, Void, Void>() {
@Override
public String tagName() {
return " my task name";
}
@Override
protected Void doInBackground(Void... voids) {
...
}
}.execute();
- ThreadPoolExecutorWithTag继承ThreadPoolExecutor，在线程的关键执行函数阶段添加了增、删Map处理
public class ThreadPoolExecutorWithTag extends ThreadPoolExecutor {
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
}
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}
public ThreadPoolExecutorWithTag(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}
//接近准确，最准确的是在afterExecute函数中删掉记录
@Override
protected void beforeExecute(Thread wt, Runnable task) {
super.beforeExecute(wt, task);
if (wt != null) {
AsyncTaskRunningInfo.removeTaskNameFromMap(wt.toString());
AsyncTaskRunningInfo.removeTaskTimeFromMap(wt.toString());
}
if (task != null) {
AsyncTaskRunningInfo.removeTaskNameFromMap(task.toString());
AsyncTaskRunningInfo.removeTaskTimeFromMap(task.toString());
}
}
@Override
public void execute(Runnable command) {
super.execute(command);
if (command == null)
throw new NullPointerException();
if (command instanceof AsyncTaskWithTag) {
AsyncTaskRunningInfo.addTaskNameToMap(command.toString(), ((AsyncTaskWithTag<?, ?, ?>) command).tagName());
}
AsyncTaskRunningInfo.addTaskTimeToMap(command.toString(), System.currentTimeMillis());
}
}
- 当自定义的线程池也遇到RejectedExecution时，可以使用ThreadPoolExecutorWithTag + AsyncTaskWithTag来实现对应分析
进阶思考
在以上流程中，需要记录对应AsyncTask的信息依赖于修改对应的Task为自定义的AsyncTaskWithTag，其中这里可以使用面向切片的编程技术，通过ASM修改字节码，根据自定义的一种规则(比如类名白名单)，把继承于AsyncTask的子类修改为继承AsyncTaskWithTag，把直接创建AsyncTask任务的代码位置修改为创建为AsyncTaskWithTag，这样可以在Gradle编译时自动替换掉AsyncTask（包括第三方库中的），可以更全面的记录到AsyncTask的创建信息和执行耗时。
