springboot中多线程批量处理数据1.配置⽂件
配置线程池参数
book:
core:
poolsize: 100
max:
poolsize: 200
queue:
capacity: 200
keepAlive:
seconds: 30
thread:
name:
prefix: zzzzz
线程池配置类
import org.springframework.beans.factory.annotation.Value;
import t.annotation.Bean;
import t.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.urrent.ThreadPoolTaskExecutor;
import urrent.ThreadPoolExecutor;
@Configuration
@EnableAsync
public class AsyncConfig {
//接收报⽂核⼼线程数
@Value("${poolsize}")
private int bookCorePoolSize;
//接收报⽂最⼤线程数
@Value("${book.max.poolsize}")
private int bookMaxPoolSize;
//接收报⽂队列容量
@Value("${book.queue.capacity}")
private int bookQueueCapacity;
/
/接收报⽂线程活跃时间（秒）
springboot推荐算法@Value("${book.keepAlive.seconds}")
private int bookKeepAliveSeconds;
//接收报⽂默认线程名称
@Value("${book.thread.name.prefix}")
private String bookThreadNamePrefix;
/**
* bookTaskExecutor:(接⼝的线程池). <br/>
*
* @return TaskExecutor taskExecutor接⼝
* @since JDK 1.8
*/
@Bean(name = "BookTask")
public ThreadPoolTaskExecutor bookTaskExecutor() {
//newFixedThreadPool
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 设置核⼼线程数
executor.setCorePoolSize(bookCorePoolSize);
// 设置最⼤线程数
executor.setMaxPoolSize(bookMaxPoolSize);
// 设置队列容量
executor.setQueueCapacity(bookQueueCapacity);
/
/ 设置线程活跃时间（秒）
executor.setKeepAliveSeconds(bookKeepAliveSeconds);
// 设置默认线程名称
executor.setThreadNamePrefix(bookThreadNamePrefix);
// 设置拒绝策略
// rejection-policy：当pool已经达到max size的时候，如何处理新任务
// CALLER_RUNS：不在新线程中执⾏任务，⽽是由调⽤者所在的线程来执⾏
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  // 等待所有任务结束后再关闭线程池
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.initialize();
return executor;
}
}
异步处理
ample.demo.dao.UserDao;
ity.UserDomain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;
import java.util.List;
import urrent.Future;
@Component
public class SyncBookHandler {
@Autowired
private UserDao userDao;
private static final Logger LOG = Logger(SyncBookHandler.class);
/**
* syncMargePsr:(多线程同步处理数据⽅法). <br/>
*
* @param bookList  ⼀段数据集合
* @param pageIndex 段数
* @return Future<String> future对象
* @author LW
* @since JDK 1.8
*/
@Async(value = "BookTask")
public Future <String> syncMargePsr(List <UserDomain> bookList, int pageIndex) {
System.out.println("thread name " + Thread.currentThread().getName());
LOG.info(String.format("此批数据的段数为:%s 此段数据的数据条数为:%s", pageIndex, bookList.size()));
//声明future对象
Future <String> result = new AsyncResult <String>("");
//循环遍历该段旅客集合
if (null != bookList && bookList.size() > 0) {
for (UserDomain book : bookList) {
try {
//数据⼊库操作
userDao.insert(book);
} catch (Exception e) {
//记录出现异常的时间，线程name
result = new AsyncResult <String>("fail,time=" + System.currentTimeMillis() + ",thread id=" + Thread.currentThread().getName() + ",pageIndex=" + pageIndex)    continue;
}
}
}
return result;
}
}
测试数据⽅法
public List <UserDomain> getPsrList() {
List <UserDomain> psrList = new ArrayList <UserDomain>();        for (int i = 0; i < 400000; i++) {
UserDomain book = new UserDomain();
book.setUserName("zcl" + i);
psrList.add(book);
}
return psrList;
}
切割数据
public void receiveBookJobRun() {
List <UserDomain> bookList = null;
bookList = getPsrList();
//⼊库开始时间
Long inserOrUpdateBegin = System.currentTimeMillis();
log.info("数据更新开始时间:" + inserOrUpdateBegin);
//接收集合各段的执⾏的返回结果
List <Future <String>> futureList = new ArrayList <Future <String>>();
/
/集合总条数
if (bookList != null) {
int listSize = bookList.size();
int listStart, listEnd;
//当总条数不⾜threadSum条时⽤总条数当做线程切分值
if (threadSum > listSize) {
threadSum = listSize;
}
//将list 切分多份多线程执⾏
for (int i = 0; i < threadSum; i++) {
//计算切割开始和结束
listStart = listSize / threadSum * i;
listEnd = listSize / threadSum * (i + 1);
//最后⼀段线程会出现与其他线程不等的情况
if (i == threadSum - 1) {
listEnd = listSize;
}
//数据切断
List <UserDomain> sunList = bookList.subList(listStart, listEnd);
//每段数据集合并⾏⼊库
futureList.add(syncBookHandler.syncMargePsr(sunList, i));
}
/
/对各个线程段结果进⾏解析
for (Future <String> future : futureList) {
String str;
if (null != future) {
try {
str = ().toString();
log.info("current thread id =" + Thread.currentThread().getName() + ",result=" + str);
} catch (InterruptedException | ExecutionException e) {
log.info("线程运⾏异常！");
}
} else {
log.info("线程运⾏异常！");
}
}
}
Long inserOrUpdateEnd = System.currentTimeMillis();
log.info("数据更新结束时间:" + inserOrUpdateEnd + "。此次更新数据花费时间为：" + (inserOrUpdateEnd - inserOrUpdateBegin));    }

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。