Android 线程池

2017-09-10  本文已影响0人  有没有口罩给我一个

在使用线程处理异步任务时,如果每次执行任务都去创建线程执行完成任务又销毁线程,由于创建线程和销毁线程会需要一些CPU资源,所以我们不得不使用线程池管理线程。

ThreadPollExecutor

ThreadPoolExecutor是创建一个线程池的类,它有四个构造方法,我们看参数最多的那个:

 public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();

    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

解释一些这些参数的作用:

*threadFactory:线程工厂,创建线程用的;

看一下执行逻辑图:


线程池执行过程.png

没有代码,一切都是纸老虎

public class ThreadDemo {
//表示执行饱和策略时退出不在执行任务
static boolean isBreak = false;

public static void main(String[] args) throws ExecutionException, InterruptedException {
    //创建线程池
    ThreadPoolExecutor executor =
            new ThreadPoolExecutor(5, 8, 60, TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(10), new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    System.out.println("饱和策略" + r);
                    isBreak = true;
                }
            });

    while (!isBreak) {
        System.out.println("核心线程数: " + executor.getCorePoolSize()
                + ", 线程池最大线程数: " + executor.getMaximumPoolSize()
                + ", 当前线程数: " + executor.getPoolSize()
                + ", 任务队列大小 " + executor.getQueue().size());

        TimeUnit.MILLISECONDS.sleep(1000);
        executor.execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
}

}

最后输出的日志:

核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 0, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 1, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 2, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 1
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 2
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 3
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 4, 任务队列大小 3
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 5, 任务队列大小 3
饱和策略RejectedExecutionException

在提交任务时会判断,如果当前执行任务的线程数少于corePoolSize,会创建新的线程来处理该任务,如果等于或多于corePoolSize,则不会创建线程,而是将该任务加入任务队列等待执行,那如果任务队列满了并且线程池的线程数量小于maximumPoolSize ,那么会创建非核心线程执行该任务,如果队列满了并且线程池的线程数量大于等于maximumPoolSize,那么会执行饱和策略。

在java中还给我们提供四种线程池,分别是:FixedThreadPoll、CachedThreadPool、SingleThreadPool和ScheduledThreadPool,这四种线程池,无非就是通过ThreadPoolExecutor的不同配置创建出来,具体这四中的实现想了解可以去看一下。

最后附上自己写的SimpleEventBus时封装的的线程池代码:

/**
   * 线程切换
 */
public class ScheduleRouterExecutor extends ThreadPoolExecutor {
//主线程
private final Handler mainHandler = new Handler(Looper.getMainLooper());
//CPU数量
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
//核心线程数量
private static final int INIT_THREAD_COUNT = CPU_COUNT + 1;
//线程池最大县城内容量
private static final int MAX_THREAD_COUNT = INIT_THREAD_COUNT * 2;
//非核心线程闲置存活的时间(秒)
private static final long SURPLUS_THREAD_LIFE = 30L;

//默认线程工厂
private static ThreadFactory DEFAULT_THREAD_FACTORY = new DefaultThreadFactory();

private static RejectedExecutionHandler REJECTED_HANDLER = new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        Log.e("tag", "提交的任务太多了哥们 ,已经执行饱和策略!");
    }
};

private static ScheduleRouterExecutor instance;

@NonNull
public static ScheduleRouterExecutor getInstance() {
    if (null == instance) {
        synchronized (ScheduleRouterExecutor.class) {
            if (null == instance) {
                instance = new ScheduleRouterExecutor(INIT_THREAD_COUNT, MAX_THREAD_COUNT,
                        SURPLUS_THREAD_LIFE, TimeUnit.SECONDS,
                        new LinkedBlockingQueue<Runnable>(64),
                        DEFAULT_THREAD_FACTORY,
                        REJECTED_HANDLER);
            }
        }
    }
    return instance;
}


private ScheduleRouterExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                               BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                               RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}


public void executeTask(@NonNull EventTask task) {
    if (task.getThreadMode() == ThreadMode.BACKGROUND) {
        if (Looper.myLooper() == Looper.getMainLooper()) {
            execute(task);
            return;
        }
        task.run();
    } else if (task.threadMode == ThreadMode.MAIN) {
        if (Looper.myLooper() != Looper.getMainLooper()) {
            mainHandler.post(task);
        } else {
            task.run();
        }
    }
}


/**
 * 线程执行结束,检查是否存在的异常
 *
 * @param r the runnable that has completed
 * @param t the exception that caused termination, or null if
 */
@Override
protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    Log.e("tag", "afterExecute");
    if (t == null && r instanceof Future<?>) {
        try {
            ((Future<?>) r).get();
        } catch (CancellationException ce) {
            t = ce;
        } catch (ExecutionException ee) {
            t = ee.getCause();
        } catch (InterruptedException ie) {
            // ignore/reset
            Thread.currentThread().interrupt();
        }
    }
    if (t != null) {
        Log.e("tag", "Running task appeared exception! Thread [" +
                Thread.currentThread().getName() + "], because [" + t.getMessage() + "]\n" +
              Arrays.toString(t.getStackTrace()));
        }
    }
}

线程工厂:

/**
   * 线程池工厂类
   */
public class DefaultThreadFactory implements ThreadFactory {

//统计线程池的数量
private static final AtomicInteger poolNumber = new AtomicInteger(1);
//统计当前线程池的线程数量
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final ThreadGroup threadGroup;
private final String namePrefix;

public DefaultThreadFactory() {
    SecurityManager securityManager = System.getSecurityManager();
    threadGroup = (securityManager != null) ? securityManager.getThreadGroup() :
            Thread.currentThread().getThreadGroup();
    namePrefix = "Simple EventBus task pool No." + poolNumber.getAndIncrement() + ", thread No.";
}

@Override
public Thread newThread(@NonNull Runnable runnable) {
    String threadName = namePrefix + threadNumber.getAndIncrement();


    Log.e("tag", "Thread production, name is [" + threadName + "]");

    final Thread thread = new Thread(threadGroup, runnable, threadName, 0);

    if (thread.isDaemon()) {//设置为非后台线程
        thread.setDaemon(false);
    }


    if (thread.getPriority() != Thread.NORM_PRIORITY) {//优先级为normal
        thread.setPriority(Thread.NORM_PRIORITY);
    }


    // 捕获多线程处理中的异常
    thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
        @Override
        public void uncaughtException(Thread thread, Throwable ex) {
            Log.e("tag", "Running task appeared exception! Thread [" + thread.getName() + "], because [" + ex.getMessage() + "]");
        }
    });
    return thread;
}
}
上一篇 下一篇

猜你喜欢

热点阅读