BY-Team
/
ad-platform
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

异步文件处理

Browse Source
master
yqy 3 years ago
parent 8ba0569459
commit 2a650960e7
10 changed files with 87 additions and 58 deletions
Show Stats Download Patch File Download Diff File

9
ad-platform-common/ad-platform-common-core/src/main/java/com/baiye/socket/filter/FeignConfiguration.java
Unescape Escape View File

@ -35,10 +35,11 @@ public class FeignConfiguration implements RequestInterceptor {
}
Map<String, String> headerMap = RequestHeaderHandler.getHeaderMap();
if (CollUtil.isNotEmpty(headerMap)) {
headerMap.forEach((headerName, headerValue) -> {
log.debug("《FeignRequestInterceptor》 headerName:【{}】 headerValue:【{}】", headerName, headerValue);
requestTemplate.header(headerName, headerValue);
});
for (Map.Entry<String, String> headerEntry : headerMap.entrySet()) {
if (headerEntry.getKey().equalsIgnoreCase(SecurityConstants.AUTHORIZATION)) {
requestTemplate.header(headerEntry.getKey(), headerEntry.getValue());
}
}
}
log.debug("========================== ↑↑↑↑↑↑ 《FeignRequestInterceptor》 End... ↑↑↑↑↑↑ ==========================");
}

7
manage/ad-platform-management/src/main/java/com/baiye/modules/system/service/impl/TaskServiceImpl.java
Unescape Escape View File

@ -73,7 +73,9 @@ public class TaskServiceImpl implements TaskService {
}
//设置默认值
task.setIsDistribution(DefaultNumberConstants.ZERO_NUMBER);
task.setCreateBy(SecurityUtils.getCurrentUserId());
if (task.getCreateBy() == null){
task.setCreateBy(SecurityUtils.getCurrentUserId());
}
taskRepository.save(task);
return new ResponseEntity<>(CommonResponse.createBySuccess(ResponseCode.SUCCESS), HttpStatus.OK);
}
@ -125,6 +127,9 @@ public class TaskServiceImpl implements TaskService {
if (task.getTotalNumber() != null){
updateTask.setTotalNumber(task.getTotalNumber());
}
if (task.getIsDistribution() != null){
updateTask.setIsDistribution(task.getIsDistribution());
}
taskRepository.save(updateTask);
}else {
throw new BadRequestException("任务id不能为空");

2
manage/ad-platform-management/src/main/java/com/baiye/timed/ReportSync.java
Unescape Escape View File

@ -52,6 +52,7 @@ public class ReportSync {
*/
@Scheduled(cron = "0 0 23 * * ? ")
public void countReport() {
log.info("++++++++++++++++++++++countReport Sync start time {} ++++++++++++++++", DateUtil.now());
String beginOfDay = DateUtil.formatDateTime(DateUtil.yesterday());
String endOfDay = DateUtil.formatDateTime(DateUtil.date());
//查询今天的线索情况
@ -63,6 +64,7 @@ public class ReportSync {
autoCountReportByTask(beginOfDay, endOfDay, callClueInfos);
//按人员统计
autoCountReportByUser(beginOfDay, endOfDay, callClueInfos);
log.info("++++++++++++++++++++++countReport Sync end time {} ++++++++++++++++", DateUtil.now());
}
private void autoCountReportByTask(String beginOfDay, String endOfDay, List<CallClueInfo> callClueInfos) {

3
services/ad-platform-source/src/main/java/com/baiye/module/dao/ClueMiddleRepository.java
Unescape Escape View File

@ -129,4 +129,7 @@ public interface ClueMiddleRepository extends JpaRepository<ClueMiddle, Long>, J
@Modifying
@Query(value = " update tb_clue_middle set member_id = ?2, member_status = 1 where clue_id in ?1 ", nativeQuery = true)
void updateMemberIdByClueIdIn(List<Long> clueIdList, Long userId);
@Query(value = "select clue_id from tb_clue_middle where task_id = ?1", nativeQuery = true)
Set<Long> findByTaskIdReturnClueId(Long taskId);
}

6
services/ad-platform-source/src/main/java/com/baiye/module/dao/ClueRecordRepository.java
Unescape Escape View File

@ -3,6 +3,8 @@ package com.baiye.module.dao;
import com.baiye.module.entity.ClueRecord;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.stereotype.Repository;
import java.util.List;
@ -23,4 +25,8 @@ public interface ClueRecordRepository extends JpaRepository<ClueRecord, Long>, J
* @param taskIds
*/
void deleteByTaskIdIn(Set<Long> taskIds);
@Modifying
@Query(value = " update tb_clue_record set status = 0 where task_id = ?1 ", nativeQuery = true)
void updateType(Long taskId);
}

27
services/ad-platform-source/src/main/java/com/baiye/module/service/impl/ClueServiceImpl.java
Unescape Escape View File

@ -60,15 +60,19 @@ public class ClueServiceImpl implements ClueService {
@Transactional(rollbackFor = Exception.class)
public void saveClue(List<Clue> clueList, Long taskId) {
if (clueList.size() > 0) {
List<Clue> clueListAll = clueRepository.saveAll(clueList);
//插入中间表数据
for (Clue clue : clueListAll) {
ClueMiddle clueMiddle = new ClueMiddle();
clueMiddle.setClueId(clue.getId());
clueMiddle.setTaskId(taskId);
clueMiddle.setMemberStatus(DefaultNumberConstants.ZERO_NUMBER);
clueMiddle.setOptimisticVersion(DefaultNumberConstants.ZERO_NUMBER);
clueMiddleRepository.save(clueMiddle);
try {
List<Clue> clueListAll = clueRepository.saveAll(clueList);
//插入中间表数据
for (Clue clue : clueListAll) {
ClueMiddle clueMiddle = new ClueMiddle();
clueMiddle.setClueId(clue.getId());
clueMiddle.setTaskId(taskId);
clueMiddle.setMemberStatus(DefaultNumberConstants.ZERO_NUMBER);
clueMiddle.setOptimisticVersion(DefaultNumberConstants.ZERO_NUMBER);
clueMiddleRepository.save(clueMiddle);
}
}catch (Exception e){
throw new BadRequestException("===================插入资源发生错误===================");
}
}
}
@ -184,14 +188,14 @@ public class ClueServiceImpl implements ClueService {
Map<String, List<String>> mapList = organizeClient.getLabel(clueDto).getBody();
List<String> organizeLabel = mapList.get("organizeLabel");
List<String> baseLabel = mapList.get("baseLabel");
clueDto.setBaseLabel(baseLabel);
clueDto.setOrganizeLabel(organizeLabel);
if (CollUtil.isEmpty(organizeLabel)){
clueDto.setOrganizeLabel(new ArrayList<>());
}
if (CollUtil.isEmpty(baseLabel)){
clueDto.setBaseLabel(new ArrayList<>());
}
clueDto.setBaseLabel(baseLabel);
clueDto.setOrganizeLabel(organizeLabel);
}
}
return clueDtoList;
@ -350,7 +354,6 @@ public class ClueServiceImpl implements ClueService {
task.setTaskName(taskName);
task.setTotalNumber(taskNum);
task.setTaskType(DefaultNumberConstants.ZERO_NUMBER);
taskClient.saveTask(task);
try {
taskClient.saveTask(task);
} catch (Exception e) {

4
services/ad-platform-source/src/main/java/com/baiye/module/service/impl/UploadFileServiceImpl.java
Unescape Escape View File

@ -122,9 +122,8 @@ public class UploadFileServiceImpl implements UploadFileService {
ClueRecord clueRecord = saveClueRecord(name, newFileUrl, uploadType, userId, taskId);
clueRecordList.add(clueRecord);
}
//异步读取文件并创建任务
fileAnalysisTask.runFileAnalysisTask(clueRecordList, oneFileName, RequestContextUtil.getHeaderMap());
fileAnalysisTask.runFileAnalysisTask(clueRecordList, oneFileName);
return new ResponseEntity<>(CommonResponse.createBySuccess(ResponseCode.SUCCESS), HttpStatus.OK);
}
@ -166,6 +165,7 @@ public class UploadFileServiceImpl implements UploadFileService {
Task task = new Task();
task.setId(taskId);
task.setTotalNumber(taskNum);
task.setIsDistribution(1);
taskClient.updateTask(task);
}
return new ResponseEntity<>(CommonResponse.createBySuccess(ResponseCode.SUCCESS), HttpStatus.OK);

79
services/ad-platform-source/src/main/java/com/baiye/task/FileAnalysisTask.java
Unescape Escape View File

@ -1,9 +1,13 @@
package com.baiye.task;
import cn.hutool.core.collection.CollUtil;
import com.alibaba.excel.EasyExcelFactory;
import com.baiye.exception.BadRequestException;
import com.baiye.feign.TaskClient;
import com.baiye.module.constant.FileConstant;
import com.baiye.module.dao.ClueJpa;
import com.baiye.module.dao.ClueMiddleRepository;
import com.baiye.module.dao.ClueRecordRepository;
import com.baiye.module.dao.ClueRepository;
import com.baiye.module.entity.Clue;
import com.baiye.module.entity.ClueRecord;
@ -24,7 +28,9 @@ import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.*;
import java.util.stream.Collectors;
@ -35,56 +41,59 @@ public class FileAnalysisTask {
@Resource
private ClueService clueService;
@Resource
private ClueJpa clueJpa;
@Resource
private ClueRepository clueRepository;
@Resource
private ClueMiddleRepository clueMiddleRepository;
@Resource
private TaskClient taskClient;
/**
* 读取文件处理数据
*
* @param clueRecords
* @param headerMap
*/
@Transactional(rollbackFor = Exception.class)
@Async(value = "SendBigDataTaskExecutor")
@SneakyThrows //处理异常try
public void runFileAnalysisTask(List<ClueRecord> clueRecords, String oneFileName, Map<String, String> headerMap) {
for (ClueRecord clueRecord : clueRecords) {
BaseExcelVo baseExcelVo = new BaseExcelVo();
baseExcelVo.setUserId(clueRecord.getCreateBy());
baseExcelVo.setClueRecordId(clueRecord.getId());
baseExcelVo.setOrigin(clueRecord.getUploadType());
baseExcelVo.setTaskId(clueRecord.getTaskId());
baseExcelVo.setOneFileName(oneFileName);
//@SneakyThrows
public void runFileAnalysisTask(List<ClueRecord> clueRecords, String oneFileName){
try {
for (ClueRecord clueRecord : clueRecords) {
BaseExcelVo baseExcelVo = new BaseExcelVo();
baseExcelVo.setUserId(clueRecord.getCreateBy());
baseExcelVo.setClueRecordId(clueRecord.getId());
baseExcelVo.setOrigin(clueRecord.getUploadType());
baseExcelVo.setTaskId(clueRecord.getTaskId());
baseExcelVo.setOneFileName(oneFileName);
FileInputStream file = new FileInputStream(clueRecord.getNewFileUrl());
switch (clueRecord.getUploadType()){
case FileConstant.ZERO_NUMBER:
log.info("==============开始导入自定义表单============{}",Thread.currentThread().getName());
EasyExcelFactory.read(file, ZdyVo.class, new ZdyExcelListener(clueService,baseExcelVo)).build().readAll();
break;
case FileConstant.ONE_NUMBER:
log.info("==============开始导入UC表单============{}",Thread.currentThread().getName());
new UcCsvListener(clueService,baseExcelVo).readUcCsv(clueRecord.getNewFileUrl());
break;
case FileConstant.TWO_NUMBER:
log.info("==============开始导入快手表单============{}",Thread.currentThread().getName());
EasyExcelFactory.read(file, KsVo.class, new KsExcelListener(clueService,baseExcelVo)).build().readAll();
break;
default:
break;
FileInputStream file = new FileInputStream(clueRecord.getNewFileUrl());
switch (clueRecord.getUploadType()) {
case FileConstant.ZERO_NUMBER:
log.info("==============开始导入自定义表单============{}", Thread.currentThread().getName());
EasyExcelFactory.read(file, ZdyVo.class, new ZdyExcelListener(clueService, baseExcelVo)).build().readAll();
break;
case FileConstant.ONE_NUMBER:
log.info("==============开始导入UC表单============{}", Thread.currentThread().getName());
new UcCsvListener(clueService, baseExcelVo).readUcCsv(clueRecord.getNewFileUrl());
break;
case FileConstant.TWO_NUMBER:
log.info("==============开始导入快手表单============{}", Thread.currentThread().getName());
EasyExcelFactory.read(file, KsVo.class, new KsExcelListener(clueService, baseExcelVo)).build().readAll();
break;
default:
break;
}
}
} catch (Exception e) {
taskClient.sendErrMessage("导入资源错误,请检查数据格式后重新导入", clueRecords.get(0).getCreateBy());
throw new BadRequestException("=========读取资源错误============");
}
//创建任务
Long taskId = clueRecords.get(0).getTaskId();
Long userId = clueRecords.get(0).getCreateBy();
if (taskId != null && userId != null && StringUtils.isNotBlank(oneFileName)){
if (taskId != null && userId != null && StringUtils.isNotBlank(oneFileName)) {
// 任务内手机号去重,先去重然后创建任务(统计任务数量)
distinctNid(taskId);
// 创建任务
@ -92,24 +101,24 @@ public class FileAnalysisTask {
baseExcelVo.setUserId(userId);
baseExcelVo.setTaskId(taskId);
baseExcelVo.setOneFileName(oneFileName);
RequestHeaderHandler.setHeaderMap(headerMap);
clueService.saveTask(baseExcelVo);
}
}
/**
* 同一个任务内去重手机号
*
* @param taskId
*/
@Transactional(propagation = Propagation.REQUIRED)
public void distinctNid(Long taskId){
public void distinctNid(Long taskId) {
Set<Long> taskIds = new HashSet();
taskIds.add(taskId);
List<Clue> clueList = clueJpa.findClue(taskIds,null);
List<Clue> clueList = clueJpa.findClue(taskIds, null);
Map<String, List<Clue>> map = clueList.stream().collect(Collectors.groupingBy(Clue::getNid));
Set<String> keySet = map.keySet();
for (String key : keySet) {
if (map.get(key).size() > 1){
if (map.get(key).size() > 1) {
List<Clue> clues = map.get(key);
clues.remove(0);
Set<Long> clueIdSet = new HashSet<>();

4
services/ad-platform-source/src/main/java/com/baiye/util/TestingUtil.java
Unescape Escape View File

@ -33,7 +33,7 @@ public class TestingUtil {
String zdyAmount = String.valueOf(list.get(3));
String zdyCollectTime = String.valueOf(list.get(4));
String zdyAddress = String.valueOf(list.get(5));
if (!(zdyName.equals("姓名") && zdyNid.equals("电话") && zdyWx.equals("微信") && zdyCollectTime.equals("日期")
if (!(zdyName.equals("姓名") && zdyNid.equals("电话") && zdyWx.equals("微信") && zdyCollectTime.equals("日期(****/**/** 00:00:00)")
&& zdyAddress.equals("线索归属地") && zdyAmount.equals("金额"))) {
log.info("======================自定义单内容格式不正确=====================");
return false;
@ -65,7 +65,7 @@ public class TestingUtil {
String taskName = String.valueOf(list.get(0));
String taskNid = String.valueOf(list.get(1));
String taskCollectTime = String.valueOf(list.get(2));
if (!(taskName.equals("姓名") && taskNid.equals("电话") && taskCollectTime.equals("日期"))) {
if (!(taskName.equals("姓名") && taskNid.equals("电话") && taskCollectTime.equals("日期(****/**/** 00:00:00)"))) {
log.info("======================自定义单内容格式不正确=====================");
return false;
}

4
services/ad-platform-source/src/main/resources/application-test.yml
Unescape Escape View File

@ -63,8 +63,8 @@ spring:
storage:
url: /usr/local/webapp/source/files/
de_symbol: /
download-template: 118.178.137.129/api-source/download/custom.xlsx
download-template-move: 118.178.137.129/api-source/download/dynamictemplate.xlsx
download-template: http://118.178.137.129:8001/source/download/custom.xlsx
download-template-move: http://118.178.137.129:8001/source/download/dynamictemplate.xlsx
# 线程池配置
save:

Write Preview
Loading…
Cancel
Save