在數(shù)據(jù)處理中,多線程用到的場景很多,在滿足計算機CPU處理能力的情況下,使用多線程可以明顯提高程序運行效率,縮短大數(shù)據(jù)處理的能力。作為java程序開發(fā),離不開spring,那么在spring中怎么創(chuàng)建多線程并將注冊到spring的類在多線程中使用呢?我自己總結(jié)了一下,可以有兩種方式,使用線程池和spring自帶多線程注解使用。
使用線程池
我一般使用固定線程數(shù)量的線程池,假如數(shù)據(jù)量很大,我會將數(shù)據(jù)放到一個大集合中,然后按照一定的比例分配數(shù)目,同時我自己寫了一個分頁類,線程的數(shù)量可以根據(jù)分頁類來自動調(diào)整。看代碼:
分頁類
package com.yiche.tag.model;
import java.util.List;
public class DataPage {
private int page = 1; // 當(dāng)前頁
public int totalPages = 0; // 總頁數(shù)
private int pageRecorders;// 每頁5條數(shù)據(jù)
private int totalRows = 0; // 總數(shù)據(jù)數(shù)
private int pageStartRow = 0;// 每頁的起始數(shù)
private int pageEndRow = 0; // 每頁顯示數(shù)據(jù)的終止數(shù)
private boolean hasNextPage = false; // 是否有下一頁
private boolean hasPreviousPage = false; // 是否有前一頁
private List list;
// private Iterator it;
public DataPage(List list, int pageRecorders) {
init(list, pageRecorders);// 通過對象集,記錄總數(shù)劃分
}
/** *//**
* 初始化list,并告之該list每頁的記錄數(shù)
* @param list
* @param pageRecorders
*/
public void init(List list, int pageRecorders) {
this.pageRecorders = pageRecorders;
this.list = list;
totalRows = list.size();
// it = list.iterator();
hasPreviousPage = false;
if ((totalRows % pageRecorders) == 0) {
totalPages = totalRows / pageRecorders;
} else {
totalPages = totalRows / pageRecorders + 1;
}
if (page >= totalPages) {
hasNextPage = false;
} else {
hasNextPage = true;
}
if (totalRows < pageRecorders) {
this.pageStartRow = 0;
this.pageEndRow = totalRows;
} else {
this.pageStartRow = 0;
this.pageEndRow = pageRecorders;
}
}
// 判斷要不要分頁
public boolean isNext() {
return list.size() > 5;
}
public void setHasPreviousPage(boolean hasPreviousPage) {
this.hasPreviousPage = hasPreviousPage;
}
public String toString(int temp) {
String str = Integer.toString(temp);
return str;
}
public void description() {
String description = "共有數(shù)據(jù)數(shù):" + this.getTotalRows() +
"共有頁數(shù): " + this.getTotalPages() +
"當(dāng)前頁數(shù)為:" + this.getPage() +
" 是否有前一頁: " + this.isHasPreviousPage() +
" 是否有下一頁:" + this.isHasNextPage() +
" 開始行數(shù):" + this.getPageStartRow() +
" 終止行數(shù):" + this.getPageEndRow();
System.out.println(description);
}
public List getNextPage() {
page = page + 1;
disposePage();
System.out.println("用戶凋用的是第" + page + "頁");
this.description();
return getObjects(page);
}
/** *//**
* 處理分頁
*/
private void disposePage() {
if (page == 0) {
page = 1;
}
if ((page - 1) > 0) {
hasPreviousPage = true;
} else {
hasPreviousPage = false;
}
if (page >= totalPages) {
hasNextPage = false;
} else {
hasNextPage = true;
}
}
public List getPreviousPage() {
page = page - 1;
if ((page - 1) > 0) {
hasPreviousPage = true;
} else {
hasPreviousPage = false;
}
if (page >= totalPages) {
hasNextPage = false;
} else {
hasNextPage = true;
}
this.description();
return getObjects(page);
}
/** *//**
* 獲取第幾頁的內(nèi)容
*
* @param page 從1開始
* @return
*/
public List getObjects(int page) {
if (page == 0)
this.setPage(1);
else
this.setPage(page);
this.disposePage();
if (page * pageRecorders < totalRows) {// 判斷是否為最后一頁
pageEndRow = page * pageRecorders;
pageStartRow = pageEndRow - pageRecorders;
} else {
pageEndRow = totalRows;
pageStartRow = pageRecorders * (totalPages - 1);
}
List objects = null;
if (!list.isEmpty()) {
objects = list.subList(pageStartRow, pageEndRow);
}
//this.description();
return objects;
}
public List getFistPage() {
if (this.isNext()) {
return list.subList(0, pageRecorders);
} else {
return list;
}
}
public boolean isHasNextPage() {
return hasNextPage;
}
public void setHasNextPage(boolean hasNextPage) {
this.hasNextPage = hasNextPage;
}
public List getList() {
return list;
}
public void setList(List list) {
this.list = list;
}
public int getPage() {
return page;
}
public void setPage(int page) {
this.page = page;
}
public int getPageEndRow() {
return pageEndRow;
}
public void setPageEndRow(int pageEndRow) {
this.pageEndRow = pageEndRow;
}
public int getPageRecorders() {
return pageRecorders;
}
public void setPageRecorders(int pageRecorders) {
this.pageRecorders = pageRecorders;
}
public int getPageStartRow() {
return pageStartRow;
}
public void setPageStartRow(int pageStartRow) {
this.pageStartRow = pageStartRow;
}
public int getTotalPages() {
return totalPages;
}
public void setTotalPages(int totalPages) {
this.totalPages = totalPages;
}
public int getTotalRows() {
return totalRows;
}
public void setTotalRows(int totalRows) {
this.totalRows = totalRows;
}
public boolean isHasPreviousPage() {
return hasPreviousPage;
}
}
這個分頁類可以多個項目復(fù)用,既可以實現(xiàn)分頁,還可以用到線程池中給線程分配數(shù)量使用。以下代碼為如何使用該分頁類給線程池使用的。
List<Integer> carModelIds = carsTagService.getAllCarModelIds();
DataPage dataPage = new DataPage(carModelIds, 300);
ExecutorService executorService = Executors.newFixedThreadPool(dataPage.totalPages);
for (int i = 0; i < dataPage.totalPages; i++) {
final int index = i + 1;
executorService.execute(new Runnable() {
List<Integer> temp = dataPage.getObjects(index);
@Override
public void run() {
try {
if (temp != null && temp.size() > 0) {
temp=Collections.synchronizedList(temp);
int sum = 0;
for (Integer carId : temp) {
try {
carsTagService.insertCarTagByCarId(carId);
sum++;
if (sum % 100 == 0) {
System.out.println("線程" + index + "當(dāng)前執(zhí)行了" + sum + "條,當(dāng)前進度是" + sum * 100/ temp.size() + "%");
}
} catch (Exception ex) {
ex.printStackTrace();
}
}
System.out.println("線程" + index + "執(zhí)行CarTagInitTask任務(wù)成功:" + temp.size());
} else {
System.out.println("CarTagInitTask任務(wù)執(zhí)行完成:未讀取到數(shù)據(jù)!");
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
executorService.shutdown();
while (true) {
if (executorService.isTerminated()) {
System.out.println("所有的子線程都結(jié)束了!");
break;
}
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
上述代碼中carModelIds就是要處理的數(shù)據(jù)集合,數(shù)目很大,如果直接for循環(huán)將會消耗很多的時間,利用線程池可以解決這個問題。但是如果直接創(chuàng)建多線程,線程中使用的對象需要final修飾,這對于spring管理的類不適用。使用線程池可以解決這個問題。
使用springboot自帶@Async注解創(chuàng)建異步線程
在springboot中,可以使用@Async注解來將一個方法設(shè)置為異步方法,調(diào)用該方法的時候,是新開一個線程去調(diào)用。代碼如下:
@Component
public class Task {
@Async
public void doTaskOne() throws Exception {
System.out.println("開始做任務(wù)一");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任務(wù)一,耗時:" + (end - start) + "毫秒");
}
@Async
public void doTaskTwo() throws Exception {
System.out.println("開始做任務(wù)二");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任務(wù)二,耗時:" + (end - start) + "毫秒");
}
@Async
public void doTaskThree() throws Exception {
System.out.println("開始做任務(wù)三");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任務(wù)三,耗時:" + (end - start) + "毫秒");
}
}
為了讓@Async注解能夠生效,還需要在Spring Boot的主程序中配置@EnableAsync,如下所示:
@SpringBootApplication
@EnableAsync
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
此時可以反復(fù)執(zhí)行單元測試,您可能會遇到各種不同的結(jié)果,比如:
- 沒有任何任務(wù)相關(guān)的輸出
- 有部分任務(wù)相關(guān)的輸出
- 亂序的任務(wù)相關(guān)的輸出
原因是目前doTaskOne、doTaskTwo、doTaskThree三個函數(shù)的時候已經(jīng)是異步執(zhí)行了。主程序在異步調(diào)用之后,主程序并不會理會這三個函數(shù)是否執(zhí)行完成了,由于沒有其他需要執(zhí)行的內(nèi)容,所以程序就自動結(jié)束了,導(dǎo)致了不完整或是沒有輸出任務(wù)相關(guān)內(nèi)容的情況。
注: @Async所修飾的函數(shù)不要定義為static類型,這樣異步調(diào)用不會生效
異步回調(diào)
為了讓doTaskOne、doTaskTwo、doTaskThree能正常結(jié)束,假設(shè)我們需要統(tǒng)計一下三個任務(wù)并發(fā)執(zhí)行共耗時多少,這就需要等到上述三個函數(shù)都完成調(diào)動之后記錄時間,并計算結(jié)果。
那么我們?nèi)绾闻袛嗌鲜鋈齻€異步調(diào)用是否已經(jīng)執(zhí)行完成呢?我們需要使用Future<T>來返回異步調(diào)用的結(jié)果,就像如下方式改造doTaskOne函數(shù):
@Async
public Future<String> doTaskOne() throws Exception {
System.out.println("開始做任務(wù)一");
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
System.out.println("完成任務(wù)一,耗時:" + (end - start) + "毫秒");
return new AsyncResult<>("任務(wù)一完成");
}
按照如上方式改造一下其他兩個異步函數(shù)之后,下面我們改造一下測試用例,讓測試在等待完成三個異步調(diào)用之后來做一些其他事情
@Test
public void test() throws Exception {
long start = System.currentTimeMillis();
Future<String> task1 = task.doTaskOne();
Future<String> task2 = task.doTaskTwo();
Future<String> task3 = task.doTaskThree();
while(true) {
if(task1.isDone() && task2.isDone() && task3.isDone()) {
// 三個任務(wù)都調(diào)用完成,退出循環(huán)等待
break;
}
Thread.sleep(1000);
}
long end = System.currentTimeMillis();
System.out.println("任務(wù)全部完成,總耗時:" + (end - start) + "毫秒");
}
看看我們做了哪些改變:
在測試用例一開始記錄開始時間
在調(diào)用三個異步函數(shù)的時候,返回Future<String>類型的結(jié)果對象
在調(diào)用完三個異步函數(shù)之后,開啟一個循環(huán),根據(jù)返回的Future<String>對象來判斷三個異步函數(shù)是否都結(jié)束了。若都結(jié)束,就結(jié)束循環(huán);若沒有都結(jié)束,就等1秒后再判斷。
跳出循環(huán)之后,根據(jù)結(jié)束時間 - 開始時間,計算出三個任務(wù)并發(fā)執(zhí)行的總耗時。
執(zhí)行一下上述的單元測試,可以看到如下結(jié)果:
開始做任務(wù)一
開始做任務(wù)二
開始做任務(wù)三
完成任務(wù)三,耗時:37毫秒
完成任務(wù)二,耗時:3661毫秒
完成任務(wù)一,耗時:7149毫秒
任務(wù)全部完成,總耗時:8025毫秒
需要注意,spring管理的異步線程數(shù)量有限,如果是web項目的話,線程數(shù)量由tomcat的線程池配置有關(guān)系,所以如果可以,最好自己配置線程配置類。
/**
* springboot里面創(chuàng)建異步線程配置類
* @author kouyy
*/
@Configuration
@EnableAsync
public class ThreadAsyncConfigurer implements AsyncConfigurer {
@Bean
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
//設(shè)置核心線程數(shù)
threadPool.setCorePoolSize(10);
//設(shè)置最大線程數(shù)
threadPool.setMaxPoolSize(100);
//線程池所使用的緩沖隊列
threadPool.setQueueCapacity(10);
//等待任務(wù)在關(guān)機時完成--表明等待所有線程執(zhí)行完
threadPool.setWaitForTasksToCompleteOnShutdown(true);
// 等待時間 (默認為0,此時立即停止),并沒等待xx秒后強制停止
threadPool.setAwaitTerminationSeconds(60);
// 線程名稱前綴
threadPool.setThreadNamePrefix("MyAsync-");
// 初始化線程
threadPool.initialize();
return threadPool;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return null;
}
}
配置類創(chuàng)建之后,以后再使用@Async創(chuàng)建異步線程就可以按照自己配置來使用了。不用擔(dān)心和spring線程池數(shù)量沖突了。