SpringBoot 多任务并行+线程池处理的实现
转载 &更新时间：日 15:57:14 & 作者：小柒
这篇文章主要介绍了SpringBoot 多任务并行+线程池处理的实现,小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
前几篇文章着重介绍了后端服务数据库和多线程并行处理优化，并示例了改造前后的伪代码逻辑。当然了，优化是无止境的，前人栽树后人乘凉。作为我们开发者来说，既然站在了巨人的肩膀上，就要写出更加优化的程序。
理论上讲，线程越多程序可能更快，但是在实际使用中我们需要考虑到线程本身的创建以及销毁的资源消耗，以及保护操作系统本身的目的。我们通常需要将线程限制在一定的范围之类，线程池就起到了这样的作用。
多任务并行+线程池处理.png
一张图能解决的问题，就应该尽可能的少BB，当然底层原理性的东西还是需要大家去记忆并理解的。
Java 线程池
Java通过Executors提供四种线程池，分别为：
newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
重用存在的线程，减少对象创建、消亡的开销，性能佳。
可有效控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。
提供定时执行、定期执行、单线程、并发数控制等功能。
方式一(CountDownLatch)
* 多任务并行+线程池统计
* 创建时间
public class StatsDemo {
final static SimpleDateFormat sdf = new SimpleDateFormat(
"yyyy-MM-dd HH:mm:ss");
final static String startTime = sdf.format(new Date());
* IO密集型任务 = 一般为2*CPU核心数（常出现于线程中：数据库数据交互、文件上传下载、网络数据传输等等）
* CPU密集型任务 = 一般为CPU核心数+1（常出现于线程中：复杂算法）
* 混合型任务 = 视机器配置和复杂度自测而定
private static int corePoolSize = Runtime.getRuntime().availableProcessors();
* public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
TimeUnit unit,BlockingQueue&Runnable& workQueue)
* corePoolSize用于指定核心线程数量
* maximumPoolSize指定最大线程数
* keepAliveTime和TimeUnit指定线程空闲后的最大存活时间
* workQueue则是线程池的缓冲队列,还未执行的线程会在队列中等待
* 监控队列长度，确保队列有界
* 不当的线程池大小会使得处理速度变慢，稳定性下降，并且导致内存泄露。如果配置的线程过少，则队列会持续变大，消耗过多内存。
* 而过多的线程又会 由于频繁的上下文切换导致整个系统的速度变缓——殊途而同归。队列的长度至关重要，它必须得是有界的，这样如果线程池不堪重负了它可以暂时拒绝掉新的请求。
* ExecutorService 默认的实现是一个无界的 LinkedBlockingQueue。
private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS,
new LinkedBlockingQueue&Runnable&(1000));
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(5);
//使用execute方法
executor.execute(new Stats("任务A", 1000, latch));
executor.execute(new Stats("任务B", 1000, latch));
executor.execute(new Stats("任务C", 1000, latch));
executor.execute(new Stats("任务D", 1000, latch));
executor.execute(new Stats("任务E", 1000, latch));
latch.await();// 等待所有人任务结束
System.out.println("所有的统计任务执行完成:" + sdf.format(new Date()));
static class Stats implements Runnable {
String statsN
CountDownL
public Stats(String statsName, int runTime, CountDownLatch latch) {
this.statsName = statsN
this.runTime = runT
this.latch =
public void run() {
System.out.println(statsName+ " do stats begin at "+ startTime);
//模拟任务执行时间
Thread.sleep(runTime);
System.out.println(statsName + " do stats complete at "+ sdf.format(new Date()));
latch.countDown();//单次任务结束，计数器减一
} catch (InterruptedException e) {
e.printStackTrace();
方式二（Future）
* 多任务并行+线程池统计
* 创建时间
public class StatsDemo {
final static SimpleDateFormat sdf = new SimpleDateFormat(
"yyyy-MM-dd HH:mm:ss");
final static String startTime = sdf.format(new Date());
* IO密集型任务 = 一般为2*CPU核心数（常出现于线程中：数据库数据交互、文件上传下载、网络数据传输等等）
* CPU密集型任务 = 一般为CPU核心数+1（常出现于线程中：复杂算法）
* 混合型任务 = 视机器配置和复杂度自测而定
private static int corePoolSize = Runtime.getRuntime().availableProcessors();
* public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
TimeUnit unit,BlockingQueue&Runnable& workQueue)
* corePoolSize用于指定核心线程数量
* maximumPoolSize指定最大线程数
* keepAliveTime和TimeUnit指定线程空闲后的最大存活时间
* workQueue则是线程池的缓冲队列,还未执行的线程会在队列中等待
* 监控队列长度，确保队列有界
* 不当的线程池大小会使得处理速度变慢，稳定性下降，并且导致内存泄露。如果配置的线程过少，则队列会持续变大，消耗过多内存。
* 而过多的线程又会 由于频繁的上下文切换导致整个系统的速度变缓——殊途而同归。队列的长度至关重要，它必须得是有界的，这样如果线程池不堪重负了它可以暂时拒绝掉新的请求。
* ExecutorService 默认的实现是一个无界的 LinkedBlockingQueue。
private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS,
new LinkedBlockingQueue&Runnable&(1000));
public static void main(String[] args) throws InterruptedException {
List&Future&String&& resultList = new ArrayList&Future&String&&();
//使用submit提交异步任务，并且获取返回值为future
resultList.add(executor.submit(new Stats("任务A", 1000)));
resultList.add(executor.submit(new Stats("任务B", 1000)));
resultList.add(executor.submit(new Stats("任务C", 1000)));
resultList.add(executor.submit(new Stats("任务D", 1000)));
resultList.add(executor.submit(new Stats("任务E", 1000)));
//遍历任务的结果
for (Future&String& fs : resultList) {
System.out.println(fs.get());//打印各个线任务执行的结果，调用future.get() 阻塞主线程，获取异步任务的返回结果
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} finally {
//启动一次顺序关闭，执行以前提交的任务，但不接受新任务。如果已经关闭，则调用没有其他作用。
executor.shutdown();
System.out.println("所有的统计任务执行完成:" + sdf.format(new Date()));
static class Stats implements Callable&String& {
String statsN
public Stats(String statsName, int runTime) {
this.statsName = statsN
this.runTime = runT
public String call() {
System.out.println(statsName+ " do stats begin at "+ startTime);
//模拟任务执行时间
Thread.sleep(runTime);
System.out.println(statsName + " do stats complete at "+ sdf.format(new Date()));
} catch (InterruptedException e) {
e.printStackTrace();
return call();
以上代码，均是伪代码，下面是2000+个学生的真实测试记录。
17:42:29.284 INFO&& 测试记录81e51ab031eb4ada92743ddf66528d82-单线程顺序执行，花费时间:3797
17:42:31.452 INFO&& 测试记录81e51ab031eb4ada92743ddf66528d82-多线程并行任务，花费时间:2167
17:42:33.170 INFO&& 测试记录81e51ab031eb4ada92743ddf66528d82-多线程并行任务+线程池，花费时间:1717
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。
您可能感兴趣的文章:
大家感兴趣的内容
12345678910
最近更新的内容
常用在线小工具还没有评论留言，赶紧来抢楼吧~~
Erlo大厅()
* 这里是“Erlo大厅”，在这发言所有人都可以看到。只保留当天信息
Erlo.vip 21:10:07Hello、欢迎使用Erlo大厅，这里是个吐槽的地方。
回车键发送
关闭极速游戏：贪吃蛇v1.0最近，由于项目里需要用到线程池来提高处理速度，记录一下spring的taskExecutor执行器来实现线程池。
这里省略了Service接口的定义和在applicationContext.xml文件中配置相应的bean（service）。
1、处理器实现类
package com.shine.
import java.util.L
import org.apache.commons.logging.L
import org.apache.commons.logging.LogF
import org.springframework.core.task.TaskE
import com.shine.service.S
处理器接口实现类
public class ProcessorImpl implements Processor {
// 日志对象
private static Log logger = LogFactory.getLog(ProcessorImpl.class);
private TaskExecutor taskE
// 业务接口
public void process() {
// 使用多线程处理
taskExecutor.execute(new Runnable(){
public void run() {
logger.debug("[Thread "
+ Thread.currentThread().getId()
+ " start]");
// 业务处理
service.handle(name);
logger.debug("[Thread "
+ Thread.currentThread().getId()
+ " end]");
} catch (RuntimeException e) {
logger.error("Service handle exception",e);
public void setService(Service service) {
this.service =
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskE
2、applicationContext.xml配置
线程池（执行器） --&
&task:executor id="taskExecutor" pool-size="1-4" queue-capacity="128" /&
&bean id="processor" class="com.shine.job.ProcessorImpl"&
&property name="service" ref="service" /&
&property name="taskExecutor"&
&ref bean="taskExecutor" /&
&/property&
其中：
pool-size="1-4"，表示线程池活跃的线程数为1，最大线程数为4；
queue-capacity="128"，表示任务队列的最大容量。
PS：关于taskExecutor还有一种配置bean来实现的，其配置的写法和参数与上面基本一致。
浏览: 12276 次
来自: 北京
(window.slotbydup=window.slotbydup || []).push({
id: '4773203',
container: s,
size: '200,200',
display: 'inlay-fix'博客分类：
1、配置文件中配置线程池
&!-- 异步线程池 --&
&bean id="taskExecutor"
class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"&
&!-- 核心线程数 --&
&property name="corePoolSize" value="10" /&
&!-- 最大线程数 --&
&property name="maxPoolSize" value="100" /&
&!-- 队列最大长度 &=mainExecutor.maxSize --&
&property name="queueCapacity" value="1000" /&
&!-- 线程池维护线程所允许的空闲时间 --&
&property name="keepAliveSeconds" value="300" /&
&!-- 线程池对拒绝任务(无线程可用)的处理策略 --&
&property name="rejectedExecutionHandler"&
&bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /&
&/property&
corePoolSize：线程池至少有2个线程是启动的，即使是空闲的也不会关闭。
maxPoolSize：最大的线程数目，当corePoolSize繁忙时，会创建线程，启动的总的线程数不能大于maxPoolSize
queueCapacity：queueCapacity： 队列大小，当corePoolSize没有空闲线程的时候，允许queueCapacity个线程任务等待,queueCapacity队列满时，才会在corePoolSize的基础上，maxPoolSize之内进行新的线程的创建。
keepAliveSeconds： 单位毫秒，超过这个时间后会将大于corePoolSize的线程关闭
rejectedExecutionHandler： 拒绝执行任务的具体操作策略，AbortPolicy表示抛出RejectedExecutionException异常。还有其他的几种选择。CallerRunsPolicy：主线程执行该任务，执行完之后尝试添加下一个任务到线程池中，可以有效降低向线程池内添加任务的速度。
2、使用线程池的实现类
import org.springframework.beans.factory.annotation.A
import org.springframework.core.task.TaskE
public class ThreadPoolTest {
@Autowired
private TaskExecutor taskE// 线程池
//也可以用如下方式引入线程池
//@Resource(name = "taskExecutor")
//private TaskExecutor taskExecutor;
// 将创建的线程添加到线程池中
public void test() throws Exception {
for (int i = 0; i & 10; i++) {
this.taskExecutor.execute(new AppContentDataPushThread());
class AppContentDataPushThread implements Runnable {
public AppContentDataPushThread() {
public void run() {
System.out.println("执行线程");
//业务代码
浏览: 36457 次
来自: 济南
(window.slotbydup=window.slotbydup || []).push({
id: '4773203',
container: s,
size: '200,200',
display: 'inlay-fix'云栖社区(yq.aliyun.com)为您免费提供Spring文集相关的问答和话题，云栖社区以分享专业、优质、高效的技术为己任，帮助技术人快速成长与发展!}