您的位置:首页 > 其它

ThreadPoolExecutor线程池的使用方法

2019-10-19 07:05 1701 查看

ThreadPoolExecutor

ThreadPoolExecutor线程池,java提供开发框架,管理线程的创建、销毁、优化、监控等。

有4种不同的任务队列:

1.ArrayBlockingQueue:基于数组结构的任务队列。此队列按先进先出的原则对任务进行排序。

2.LinkedBlockingQueue:基于链表结构的任务队列。此队列也是按先进先出的原则对任务进行排序。但性能比ArrayBlockingQueue高。

3.synchronousQueue:不存储元素的任务队列。每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态。

4.PriorityBlockingQueue:具有优先级的任务队列。此队列中的元素必须能够比较。

拒绝策略:

RejectedExecutionHandler(饱和策略 ):当线程池中的线程数大于maximumPoolSize时,线程池就不能在处理任何任务了,这时线程池会抛出异常。原因就是这个策略默认情况下是AbortPolicy:表示无法处理新任务时抛出异常。

1.AbortPolicy:直接抛出异常。

2.CallerRunsPolicy:只用调用者所在线程来运行任务。

3.DiscardOldestPolicy:丢弃队列里最近的一个任务,并执行当前任务

4.DiscardPolicy:不处理,丢弃掉。

自定义:

ThreadPoolExecutor.AbortPolicy()
//抛出java.util.concurrent.RejectedExecutionException异常
ThreadPoolExecutor.CallerRunsPolicy()
//重试添加当前的任务,他会自动重复调用execute()方法
ThreadPoolExecutor.DiscardOldestPolicy()
//抛弃旧的任务
ThreadPoolExecutor.DiscardPolicy()
// 抛弃当前的任务

private static class RecjectThreadHandler implements RejectedExecutionHandler
{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

}

// 异常记录
private void doLog(Runnable r, ThreadPoolExecutor executor)
{
System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount());
}
}

创建线程工厂:

用来创建线程。

public class CheckThreadFactory implements ThreadFactory
{
private String threadGroupName;

private AtomicInteger count = new AtomicInteger(0);

public CheckThreadFactory(String threadGroupName) {
this.threadGroupName = threadGroupName;
}

@Override
public Thread newThread(Runnable r)
{
Thread thread = new Thread(r);
thread.setName(threadGroupName+"--"+count.addAndGet(1));
thread.setPriority(5);
thread.setDaemon(true);.// 设置为守护线程, 默认为主线程
return thread;
}
}

线程工具类:

/**
* @author Donald
* @create 2019-09-21 21:40
*/
public class CheckExcetPool
{
// 线程池核心线程数
private static int corePoolSize = Runtime.getRuntime().availableProcessors() * 5;
// 最大线程数
private static int maximumPoolSize = corePoolSize > 255 ? 255 : corePoolSize * 2;
// 线程池中除了核心线程,其他线程的最大存活时间
private static int keepAliveTime = 60;
// 时间单位
private static TimeUnit timeUnit = TimeUnit.SECONDS;
// 线程等待队列
private static BlockingQueue queue = new LinkedBlockingQueue();
//private static BlockingQueue queue = new ArrayBlockingQueue<Runnable>(30);
// 创建线程的工厂
private static CheckThreadFactory checkThreadFactory = new CheckThreadFactory("checkGroup");
// 拒绝策略 当提交任务数超过maxmumPoolSize+workQueue之和时,
// * 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),
// *   任务会交给RejectedExecutionHandler来处理
/*handler的拒绝策略:
有四种:第一种AbortPolicy:不执行新任务,直接抛出异常,提示线程池已满
第二种DisCardPolicy:不执行新任务,也不抛出异常
第三种DisCardOldSetPolicy:将消息队列中的第一个任务替换为当前新进来的任务执行
第四种CallerRunsPolicy:直接调用execute来执行当前任务*/

private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
timeUnit,
queue,
checkThreadFactory
);

public static void submit( Runnable runnable)
{
System.out.println(corePoolSize+"::"+queue.size());
threadPoolExecutor.submit(runnable);
}
public static <T> Future submit(Callable<T> callable)
{
return threadPoolExecutor.submit(callable);
}
public static <T> void excutor( Runnable run, T result )
{
threadPoolExecutor.submit( run,result );
}
public static void excutor( Runnable run)
{
threadPoolExecutor.execute( run);
}
private static class RecjectThreadHandler implements RejectedExecutionHandler
{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

}

// 异常记录
private void doLog(Runnable r, ThreadPoolExecutor executor)
{
System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount());
}
}
}

线程服务类,实现runnable 接口:

/**
* @author Donald
* @create 2019-09-21 23:00
*/
public class ThreadService implements Runnable
{
private CountDownLatch countDownLatch;
private UserInterface userInterface;

public ThreadService(CountDownLatch countDownLatch, UserInterface userInterface) {
this.countDownLatch = countDownLatch;
this.userInterface = userInterface;
}

@Override
public void run()
{
try {
long start = System.currentTimeMillis();
userInterface.doSomething();
System.err.println(String.format("user time :%s",System.currentTimeMillis()-start));
Thread.sleep(1000);
}catch ( Exception e)
{
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
}
}

具体业务逻辑:

/**
* @author Donald
* @create 2019-09-21 22:51
*/
public interface UserInterface
{
void doSomething();
}

业务类:

/**
* @author Donald
* @create 2019-09-21 22:51
*/
public class UserService implements UserInterface
{
private int number;

public UserService(int number) {
this.number = number;
}

@Override
public void doSomething() {
System.out.println(Thread.currentThread().getName()+"<<<<"+number);
}

}

以上就是本文的全部内容,希望对大家的学习有所帮助

您可能感兴趣的文章:

内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息