ThreadPoolExecutor线程池的使用方法

ThreadPoolExecutor线程池的使用方法

ThreadPoolExecutor

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

有4种不同的任务队列:

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

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

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

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

拒绝策略：

RejectedExecutionHandler（饱rFhSyaNPUX和策略 ）：当线程池中的线程数大于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;

prFhSyaNPUXrivate 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(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 Future submit(Callable callable)

{

return threadPoolExecutor.submit(callable);

}

public static 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 UserInterfahttp://ce

{

private int number;

public UserService(int number) {

this.number = number;

}

@Override

public void doSomething() {

System.out.println(Thread.currentThread().getName()+"<<<<"+number);

}

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。