异步线程池的使用

Posted 2021-11-28 gaopengfirst

合理使用异步线程开发项目能提高一个项目的并发量，减少响应时间。下面就简单介绍一下异步线程池的使用，参考博客：https://blog.csdn.net/hry2015/article/details/67640534

spring 对@Async定义异步任务的方法有3种：

1.最简单的异步调用，返回值为void；

2.带参数的异步调用，异步方法可以传入参数；

3.异常调用返回Future

代码如下：

package com.hry.spring.async.annotation;

import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

/**
 * 异步方法调用
 * 
 * @author hry
 *
 */
@Component
public class AsyncDemo {
	private static final Logger log = LoggerFactory.getLogger(AsyncDemo.class);

	/**
	 * 最简单的异步调用，返回值为void
	 */
	@Async
	public void asyncInvokeSimplest() {
		log.info("asyncSimplest");
	}

	/**
	 * 带参数的异步调用 异步方法可以传入参数
	 * 
	 * @param s
	 */
	@Async
	public void asyncInvokeWithParameter(String s) {
		log.info("asyncInvokeWithParameter, parementer={}", s);
	}

	/**
	 * 异常调用返回Future
	 * 
	 * @param i
	 * @return
	 */
	@Async
	public Future<String> asyncInvokeReturnFuture(int i) {
		log.info("asyncInvokeReturnFuture, parementer={}", i);
		Future<String> future;
		try {
			Thread.sleep(1000 * 1);
			future = new AsyncResult<String>("success:" + i);
		} catch (InterruptedException e) {
			future = new AsyncResult<String>("error");
		}
		return future;
	}

}

　　

spring启动配置的方式有两种：

1.注解；

2.XML配置文件

我们先说注解的方式：

@EnableAsync注解可以开启异步调用功能， public AsyncTaskExecutor taskExecutor() 方法自定义自己的线程池，线程池前缀”Anno-Executor”。如果不定义，则使用系统默认的线程池。代码如下：

package com.hry.spring.async.annotation;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * 通过@EnableAsync启动异步方法
 * @author hry
 * 
 */
@SpringBootApplication
@EnableAsync // 启动异步调用
public class AsyncApplicationWithAnnotation {
	private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithAnnotation.class);
	
	/**
	 * 自定义异步线程池
	 * 	如果没有这个方法，则使用默认的线程池
	 * @return
	 */
    @Bean
    public AsyncTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setThreadNamePrefix("Anno-Executor");//设置线程名称的前缀
        executor.setCorePoolSize(5);//设置核心线程数
        executor.setMaxPoolSize(10);//设置最大线程数

        // 设置拒绝策略
        executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
			@Override
			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
				// .....
			}
		});
        // 使用预定义的异常处理类
        // executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        return executor;
    }
    
	public static void main(String[] args) {
		log.info("Start AsyncApplication.. ");
		SpringApplication.run(AsyncApplicationWithAnnotation.class, args);
	}
}

　现在写测试类：

package com.hry.spring.async.annotation;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.hry.spring.async.annotation.AsyncApplicationWithAnnotation;
import com.hry.spring.async.annotation.AsyncDemo;

@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithAnnotation.class)
public class AsyncApplicationWithAnnotationTests {
	@Autowired
	private AsyncDemo asyncDemo;
	
	@Test
	public void contextLoads() throws InterruptedException, ExecutionException {
		asyncDemo.asyncInvokeSimplest();
		asyncDemo.asyncInvokeWithParameter("test");
		Future<String> future = asyncDemo.asyncInvokeReturnFuture(1000);
		System.out.println(future.get());
	}
}

　启动测试类，结果如下：

说明启动异步线程池成功，执行上面的3个方法分别用了3个线程。

再说一下XML配置文件配置的方式：

我们先创建XML配置文件：

<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop"  
    xmlns:context="http://www.springframework.org/schema/context"  
    xmlns:tx="http://www.springframework.org/schema/tx"  
    xmlns:task="http://www.springframework.org/schema/task"  
    xsi:schemaLocation="  
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
        http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd  
        http://www.springframework.org/schema/context   
http://www.springframework.org/schema/jdbc/spring-jdbc-3.0.xsd  
        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd  
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd"  
    default-lazy-init="false">  
    
  	<!-- 等价于 @EnableAsync， executor指定线程池 -->
  	<task:annotation-driven executor="xmlExecutor"/>
  	<!-- id指定线程池产生线程名称的前缀 -->
    <task:executor
        id="xmlExecutor"
        pool-size="3-20"
        queue-capacity="100"
	    keep-alive="120"
        rejection-policy="CALLER_RUNS"/>
	
</beans>

　　

其中id表示线程名称的前缀，pool-size表示线程池的大小，“3-20”表示线程池的最大线程数为20，最小线程数为3，也可以设置成一个值，如：pool-size=5，则表示线程池的核心线程数和最大线程数相同，都是5。queue-capacity表示排队队列的长度。

项目启动时读取xml配置文件，如下：

package com.hry.spring.async.xml;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ImportResource;


/**
 * 通过XML启动异步
 * @author hry
 *
 */
@SpringBootApplication
@ImportResource("classpath:/async/spring_async.xml")
public class AsyncApplicationWithXML {
	private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithXML.class);
	
	public static void main(String[] args) {
		log.info("Start AsyncApplication.. ");
		SpringApplication.run(AsyncApplicationWithXML.class, args);
	}
}　　

 

编写测试类：

package com.hry.spring.async.xml;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
@RunWith(SpringRunner.class)
@SpringBootTest(classes=AsyncApplicationWithXML.class)
public class AsyncApplicationWithXMLTest {
	@Autowired
	private AsyncDemo asyncDemo;

	@Test
	public void contextLoads() throws InterruptedException, ExecutionException {
		asyncDemo.asyncInvokeSimplest();
		asyncDemo.asyncInvokeWithParameter("test");
		Future<String> future = asyncDemo.asyncInvokeReturnFuture(100);
		System.out.println(future.get());
	}
}

　　

测试结果如下：

测试成功。