场景##
公司一站通翻译同步系统 , 需要将客户保存在美国站点的文章翻译并保存到其余22个站点的数据库。由于翻译需要耗费较长的时间,故而使用队列将任务投递到线程池中处理，
- Springboot 配置产品同步核心线程池
package cn.configuration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
@Configuration
@EnableAsync // 开启线程池
public class AsyncThreadPoolConfiguration
{
/**
* 配置默认线程池，用于处理一些公共异步任务
*/
@Bean("defaultThread")
public Executor defaultThread(){
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(20);// 核心线程数,
executor.setMaxPoolSize(40);// 并发线程的数量限制为2
executor.setQueueCapacity(200); // 线程队列
executor.setThreadNamePrefix("defaultThread@");
executor.initialize();
return executor;
}
/**
* 同步产品使用的线程池
* @return
*/
@Bean("syncProduct")
public Executor syncProduct() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(22);// 核心线程数,
executor.setMaxPoolSize(22);// 并发线程的数量限制为2
executor.setQueueCapacity(100); // 线程队列
executor.setThreadNamePrefix("syncProduct@");
executor.initialize();
return executor;
}
/**
* todo 配置其他功能的线程池
*/
}
2.1 使用方式一 注入线程池对象 ,通过 lomda表达式结合Future实现
package cn.services;
import cn.utils.SystemUtils;
import cn.utils.TimeUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;
@Service
public class IndexServices
{
// 注入线程池对象
@Resource(name = "syncProduct")
private ThreadPoolTaskExecutor syncProduct;
public void ansyc() throws InterruptedException {
// 初始化任务结果集
Map<String,Future<String>> futures = new HashMap<>();
// 模拟需要同步的 22个子站点
List<String> langs = new ArrayList<>();
for (int i=1;i<=22;i++){
langs.add("lang"+i);
}
// 多线程处理任务
for (String lang : langs) {
Future<String> future = syncProduct.submit(() -> {
// 真正的业务处理
if(lang.equals("lang10")){
Thread.sleep(8000);// 模拟阻塞时间
}else{
Thread.sleep(3000); // 模拟阻塞时间
}
System.out.println("打印结果 : " + lang + "当前线程名称是:"+SystemUtils.getThreadID());
return lang + "...";
});
futures.put(lang,future);
}
// 阻塞等待结果，缺点该种方式无法设置阻塞等待有效时间，如果有一个线程阻塞,会导致整个线程池一直等待
for (Map.Entry<String, Future<String>> entry : futures.entrySet()) {
while (true){
Future<String> future = entry.getValue();
String lang = entry.getKey();
if(future.isDone() && !future.isCancelled()){
String result = null;
try{
result = future.get();
//result = future.get(6L,TimeUnit.SECONDS); 设置阻塞等待时间无效
}catch(ExecutionException e){
e.printStackTrace();
}
System.out.println(lang+" 站点任务结果result=" + result + "获取完成!" + TimeUtils.getCurretDate());
break;
}else{
Thread.sleep(1);//每次轮询休息1毫秒（CPU纳秒级），避免CPU高速轮循耗空CPU ，这个至关重要
}
}
}
/*
// 设置阻塞时间有效版本
for (Map.Entry<String, Future<String>> entry : futures.entrySet()) {
while (true){
Future<String> future = entry.getValue();
String lang = entry.getKey();
String result = null;
try{
result = future.get(3L,TimeUnit.SECONDS);
}catch(ExecutionException e){
e.printStackTrace();
}catch (TimeoutException e){
System.out.println("存在超时的任务,无法获取结果");
}
System.out.println(lang+" 站点任务结果result=" + result + "获取完成!" + TimeUtils.getCurretDate());
break;
}
}
*/
}
}
2.2 上线方式二，通过Springboot的Ansyc注解实现任务并发
实现思路是 写一个 @Ansyc 注解标记的业务方法，在外层循环调用，通过参数控制实现不同的逻辑
/**
* 翻译英文站点文章 并同步到22个子站点
*/
// 业务核心
@Async("syncProduceToSite")
public CompletableFuture<String> syncProduceToSite(String lang) throws InterruptedException {
// 通过 参数 lang的不同，实现翻译成不同的语言，并推送到不同站点的数据库
Thread.sleep(8000L);
String results = lang + " success";
String name = SystemUtils.getCurrentThreadName();
return CompletableFuture.completedFuture("站点"+lang + "处理的结果是:"+results+";线程名称是:"+name);
}
在外层调用过程
// 模拟准备 22个站点的语言标识符参数
List<String> langs = new ArrayList<>();
for (int i=1;i<=22;++i){
langs.add("lang"+i);
}
// 初始化结果集
Map<String,CompletableFuture<String>> futureMap = new HashMap<>();
for (String lang: langs) {
// 业务执行
CompletableFuture<String> future = authServices.syncProduceToSite(lang);
futureMap.put(lang,future); // 将结果集放入 map
}
// 获取结果集
for (Map.Entry<String, CompletableFuture<String>> entry : futureMap.entrySet()){
String lang = entry.getKey();
CompletableFuture<String> future = entry.getValue();
try {
// CompletableFuture.anyOf(future).join(); // 阻塞等待结果的返回,加上这一句，那么下面的阻塞时间就无效
String result= future.get(2L, TimeUnit.SECONDS);
System.out.println("站点:"+lang+" 得到结果啦 "+result);
} catch (TimeoutException e) {
// e.printStackTrace();
System.out.println("站点:"+lang+" 超时,无法获取结果");
}
}
