一般在使用多线程的时候都会使用到线程池，好处网上有很多，最主要的好处是线程池中的线程可以重复利用，减少线程创建销毁带来的资源浪费。
java实现线程池一般有两种：
- java自带的线程池Executor；
- Spring封装的线程池。
如果使用java自带的线程池编写起来比较麻烦，找了个网络的例子：
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolTest{
public static void main(String[] args){
//创建等待队列
BlockingQueue<Runnable> bqueue = new ArrayBlockingQueue<Runnable>(20);
//创建线程池，池中保存的线程数为3，允许的最大线程数为5
ThreadPoolExecutor pool = new ThreadPoolExecutor(3,5,50,TimeUnit.MILLISECONDS,bqueue);
//创建七个任务
Runnable t1 = new MyThread();
Runnable t2 = new MyThread();
Runnable t3 = new MyThread();
Runnable t4 = new MyThread();
Runnable t5 = new MyThread();
Runnable t6 = new MyThread();
Runnable t7 = new MyThread();
//每个任务会在一个线程上执行
pool.execute(t1);
pool.execute(t2);
pool.execute(t3);
pool.execute(t4);
pool.execute(t5);
pool.execute(t6);
pool.execute(t7);
//关闭线程池
pool.shutdown();
}
}
class MyThread implements Runnable{
@Override
public void run(){
System.out.println(Thread.currentThread().getName() + "正在执行。。。");
try{
Thread.sleep(100);
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
- 首先你自定义一个queue；
- 你还要创建一个java的ThreadPoolExecutor的线程，构造方法中有一大最参数；
- 还要定义Runnable。
看着太麻烦了。不要着急Spring这些已经帮咱们做了。
一个忠告：在开发时，如果遇到通用的功能，试着到开源社区寻找，一般都能找到现成的方法或解决方案，切勿闭门造车。
来让我们看看使用Spring是怎样开发的。
<bean id="cacheExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<!-- 线程池维护线程的核心数量 -->
<property name="corePoolSize" value="2"/>
<!-- 线程池维护线程的最大数量 -->
<property name="maxPoolSize" value="4"/>
<!-- 缓存队列 -->
<property name="queueCapacity" value="64"/>
<!-- 线程池的名称 -->
<property name="threadNamePrefix" value="CACHE_DISTANCE" />
<!-- 对拒绝task的处理策略 -->
<property name="rejectedExecutionHandler">
<bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy"/>
</property>
</bean>
// 通过spring注入到需要的类中
@Autowired
ThreadPoolTaskExecutor cacheExecutor;
// 在需要使用的方法中
public void startCacheThreads() {
cacheExecutor.submit(new Runnable() {
@Override
public void run() {
try {
// 是我们需要多线程实现的方法
doCacheTask();
} catch (Exception ex) {
logger.info("更新缓存失败：{}", Thread.currentThread().getName(), ex);
}
}
});
}
是不是这么写很高效，有容易理解。
在实战开发中会用到自定义rejectedExecutionHandler，就是任务太多导致线程池中queue已经塞满的拒绝策略，spring提供了这么几个策略：
- ThreadPoolExecutor.AbortPolicy策略，是默认的策略,处理程序遭到拒绝将抛出运行时 RejectedExecutionException；
- ThreadPoolExecutor.CallerRunsPolicy策略 ,调用者的线程会执行该任务,如果执行器已关闭,则丢弃；
- ThreadPoolExecutor.DiscardPolicy策略，不能执行的任务将被丢弃；
- ThreadPoolExecutor.DiscardOldestPolicy策略，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。
但是有些情况下，我们想监控到底有多少被抛弃掉了，那么我们就需要自定义策略，怎么写呢，也很简单。
<bean id="uploadCoordinateToNearbyExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="1"/>
<property name="maxPoolSize" value="2"/>
<property name="queueCapacity" value="16384"/>
<property name="threadNamePrefix" value="UPLOADCOORDINATETONEARBY_EXECUTOR" />
<property name="rejectedExecutionHandler">
<bean class="com.yd.util.UploadToNearbyDiscardPolicy">
<constructor-arg value="uploadCoordinateToNearby"/>
</bean>
</property>
</bean>
com.yd.util.UploadToNearbyDiscardPolicy就是自定义路由,代码需要implements RejectedExecutionHandler接口。
public class UploadToNearbyDiscardPolicy implements RejectedExecutionHandler {
private static final Logger logger = LoggerFactory.getLogger(UploadToNearbyDiscardPolicy.class);
public final String name;
public long discardTimes;
public UploadToNearbyDiscardPolicy(String name) {
this.name = name;
}
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
++StatObject.instance.uploadToNearbyDiscardCount;
logger.warn("{}线程池太忙丢弃任务:{}", name, ++discardTimes);
}
}
在上面的例子中，可以添加计数器，也可以将其数量上报的监控系统中。
多说一个Guava线程池返回数据的使用方法ListenableFuture。在开发中会常常遇到多线程计算后返回数据，常见的会写一个while效率一直检查线程是否处理完成，这个也可以，不过Guava提供了一个更下高效优雅的方法。
public static void main(String[] args) {
// 将 ExecutorService 转为 ListeningExecutorService，可以使用MoreExecutors.listeningDecorator(ExecutorService)进行装饰
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
// 多线程处理并返回结果
ListenableFuture explosion = service.submit(new Callable() {
public Explosion call() {
return pushBigRedButton();
}
});
Futures.addCallback(explosion, new FutureCallback() {
// we want this handler to run immediately after we push the big red button!
public void onSuccess(Explosion explosion) {
walkAwayFrom(explosion);
}
public void onFailure(Throwable thrown) {
// 抛出异常时走这里
battleArchNemesis(); // escaped the explosion!
}
});
}
更下详尽的例子可以参考这篇blog，Guava - 并行编程Futures
网友评论