编写一个抽象类
import com.alibaba.druid.util.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.iflytek.epdcloud.mhk.mark.common.entity.MachineEvaluate;
import com.iflytek.epdcloud.mhk.mark.common.enumeration.ResponseEnumeration;
import com.iflytek.epdcloud.mhk.mark.web.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.multipart.MultipartFile;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@Slf4j
public abstract class FileUploadService {
public T t;
//默认一批是1000个
public static final IntegerbatchSizeIndex =1000;
public static final IntegerThreadNumIndex =15;
/**
* 上传文件
* @param file
* @param fileRunnable
* @param batchSize
* @param threadNum
* @param list
* @param args
*/
public void uploadFile(MultipartFile file,FileRunnable fileRunnable,Integer batchSize,Integer threadNum,ArrayList list,Object ... args){
analyse(file,fileRunnable,batchSize,threadNum,list,args);
}
/**
* 解析文件
* @param file
* @param fileRunnable
* @param batchSize
* @param threadNum
* @param list
* @param args
*/
public void analyse(MultipartFile file,FileRunnable fileRunnable,Integer batchSize,Integer threadNum,ArrayList list,Object... args){
AtomicInteger atomicInteger =new AtomicInteger(0);
HashMap map =new HashMap<>();
BufferedReader bufferedReader =null;
InputStreamReader inputStreamReader =null;
InputStream is=null;
try {
is = file.getInputStream();
inputStreamReader =new InputStreamReader(is);
bufferedReader =new BufferedReader(inputStreamReader);
String line =null;
while((line= bufferedReader.readLine())!=null) {
String[] array = line.split("\t");
List collect = Arrays.stream(array).filter(s12 ->
!StringUtils.isEmpty(s12)
).collect(Collectors.toList());
if((!CollectionUtils.isEmpty(collect))){
atomicInteger.incrementAndGet();
modify(collect,list,args);
}else {
log.error("第{}行的数据格式有误",atomicInteger.toString());
}
}
bufferedReader.close();
}catch (Exception e){
log.error("文件解析失败:{}",e.getMessage());
throw new BusinessException(ResponseEnumeration.MACHINE_FILE_ANALYSE_ERROR);
}finally {
try {
if (bufferedReader !=null) {
bufferedReader.close();
}
if (inputStreamReader !=null) {
inputStreamReader.close();
}
}catch (IOException e) {
throw new BusinessException(ResponseEnumeration.MACHINE_FILE_ANALYSE_ERROR);
}
}
batchSave(list, batchSize, fileRunnable,threadNum);
}
/**
* 批量添加数据到数据库
* @param modifyList
* @param batchSize
* @param runnable
* @param threadNum
*/
public void batchSave(List modifyList, Integer batchSize, FileRunnable runnable, Integer threadNum) {
if(CollectionUtils.isEmpty(modifyList)){
log.info("没有数据要添加数据库");
return;
}
if(ObjectUtils.isEmpty(batchSize))
batchSize =batchSizeIndex;
if(ObjectUtils.isEmpty(threadNum))
threadNum =ThreadNumIndex;
int batchNum = modifyList.size() / batchSize;
CountDownLatch countDownLatch =new CountDownLatch(batchNum+1);
LinkedBlockingQueue runnables =new LinkedBlockingQueue<>(Integer.MAX_VALUE);
ThreadPoolExecutor threadPoolExecutor =new ThreadPoolExecutor(threadNum,
threadNum,
20,
TimeUnit.SECONDS,
runnables,new BatchSaveThreadFactory());
List newList =null;
for (int i=1;i<=batchNum+1;i++
) {
if(i!=batchNum+1){
newList = modifyList.subList((i-1)*batchSize,i*batchSize);
FileRunnable fileRunnable = runnable.newInstance();
fileRunnable.setCountDownLatch(countDownLatch);
fileRunnable.setList(newList);
threadPoolExecutor.execute(fileRunnable);
}else {
newList = modifyList.subList((i-1)*batchSize,modifyList.size());
FileRunnable fileRunnable = runnable.newInstance();
fileRunnable.setCountDownLatch(countDownLatch);
fileRunnable.setList(newList);
threadPoolExecutor.execute(fileRunnable);
}
}
try {
countDownLatch.await();
}catch (InterruptedException e) {
log.error("文件落盘失败:{}",e.getMessage());
throw new BusinessException(ResponseEnumeration.MACHINE_FILE_SAVE_ERROR);
}
threadPoolExecutor.shutdown();
}
/**
* 解析之后每行数据的处理
* @param collect
* @param list
* @param args
*/
public abstract void modify(List collect,List list,Object... args);
}
具体每个线程执行的业务逻辑
在落盘时要根据业务场景选择是否去重
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
public abstract class FileRunnableimplements Runnable{
private CountDownLatchcountDownLatch;
private Listlist;
public abstract FileRunnablenewInstance();
@Override
public void run() {
long l = System.currentTimeMillis();
doSave(list);
countDownLatch.countDown();
System.out.println("耗时"+(System.currentTimeMillis()-l));
}
public abstract void doSave(List list);
public void setCountDownLatch(CountDownLatch countDownLatch) {
this.countDownLatch = countDownLatch;
}
public void setList(List list) {
this.list = list;
}
}
继承抽象类进行操作
import com.alibaba.druid.util.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.iflytek.epdcloud.mhk.mark.common.entity.MachineEvaluate;
import com.iflytek.epdcloud.mhk.mark.common.enumeration.MachineEvaluateStatusEnumeration;
import com.iflytek.epdcloud.mhk.mark.common.enumeration.ResponseEnumeration;
import com.iflytek.epdcloud.mhk.mark.common.service.IMachineEvaluateService;
import com.iflytek.epdcloud.mhk.mark.common.utils.SnowflakeIdWorker;
import com.iflytek.epdcloud.mhk.mark.web.buiness.MachineEvaluateBusiness;
import com.iflytek.epdcloud.mhk.mark.web.exception.BusinessException;
import com.iflytek.epdcloud.mhk.mark.web.executor.BatchSaveThreadFactory;
import com.iflytek.epdcloud.mhk.mark.web.executor.FileUploadService;
import com.iflytek.epdcloud.mhk.mark.web.executor.MachineEvalutteTask;
import com.iflytek.epdcloud.mhk.mark.web.model.param.MachineEvaluateDto;
import com.iflytek.epdcloud.mhk.mark.web.model.vo.MachineEvaluateVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import java.util.*;
/**
*
*/
@Slf4j
@Service
public class MachineEvaluateBusinessImplextends FileUploadServiceimplements MachineEvaluateBusiness {
@Autowired
private IMachineEvaluateServiceiMachineEvaluateService;
@Transactional
@Override
public void upload(MachineEvaluateDto dto) {
HashMap map =new HashMap<>();
ArrayList machineEvaluateArrayList =new ArrayList<>();
MachineEvalutteTask machineEvalutteTask =new MachineEvalutteTask(iMachineEvaluateService, machineEvaluateArrayList);
uploadFile(dto.getFile(),machineEvalutteTask,null ,null,machineEvaluateArrayList,dto.getQuestionTypeId(),map);
}
@Override
public MachineEvaluateVogetUploadResult(Long questionTypeId) {
MachineEvaluateVo machineEvaluateVo =new MachineEvaluateVo();
Long total =iMachineEvaluateService.getTotal(questionTypeId,null);
Long validTotal =iMachineEvaluateService.getTotal(questionTypeId,MachineEvaluateStatusEnumeration.EVALUATE_ENTERED.getCode());
machineEvaluateVo.setUploadTotal(total);
machineEvaluateVo.setValidTotal(validTotal);
return machineEvaluateVo;
}
@Override
public void modify(List collect,List list, Object... args) {
{
while((collect.size()>=2)&&(collect.size()%2==0))
{
List strings = collect.subList(0, 2);
String s1 = strings.get(0);
String s2 = strings.get(1);
Long questionTypeId = (Long) args[0];
Map map = (Map) args[1];
strings.clear();
String s3 = map.get(s1+questionTypeId);
if(!StringUtils.isEmpty(s3))
break;
map.put(s1+questionTypeId,s2);
MachineEvaluate machineEvaluate =new MachineEvaluate();
machineEvaluate.setExamId(s1);
machineEvaluate.setQuestionTypeId(questionTypeId);
machineEvaluate.setScore(s2);
machineEvaluate.setStatus(MachineEvaluateStatusEnumeration.EVALUATE_NOT_ENTER.getCode());
machineEvaluate.setId(SnowflakeIdWorker.generateId());
list.add(machineEvaluate);
}
}
}
}
