|
|
@ -14,6 +14,7 @@ import me.zhengjie.modules.buildrecord.service.dto.BuildRecordDto;
|
|
|
|
import me.zhengjie.modules.buildrecord.task.dto.SendBigDataDTO;
|
|
|
|
import me.zhengjie.modules.buildrecord.task.dto.SendBigDataDTO;
|
|
|
|
import me.zhengjie.modules.remoterec.domain.RemoteRecord;
|
|
|
|
import me.zhengjie.modules.remoterec.domain.RemoteRecord;
|
|
|
|
import me.zhengjie.modules.remoterec.service.RemoteRecordService;
|
|
|
|
import me.zhengjie.modules.remoterec.service.RemoteRecordService;
|
|
|
|
|
|
|
|
import me.zhengjie.modules.remoterec.service.dto.RemoteRecordDto;
|
|
|
|
import me.zhengjie.modules.tag.domain.Tag;
|
|
|
|
import me.zhengjie.modules.tag.domain.Tag;
|
|
|
|
import me.zhengjie.modules.tag.service.TagService;
|
|
|
|
import me.zhengjie.modules.tag.service.TagService;
|
|
|
|
import me.zhengjie.modules.tag.service.dto.TagDto;
|
|
|
|
import me.zhengjie.modules.tag.service.dto.TagDto;
|
|
|
@ -101,23 +102,18 @@ public class SendBigDataTask {
|
|
|
|
|
|
|
|
|
|
|
|
private void runTask(BuildRecord resource, RemoteRecord remoteRecord, SendBigDataDTO sendBigDataDTO) {
|
|
|
|
private void runTask(BuildRecord resource, RemoteRecord remoteRecord, SendBigDataDTO sendBigDataDTO) {
|
|
|
|
|
|
|
|
|
|
|
|
// 根据发送任务的Id来读取发送号码表
|
|
|
|
|
|
|
|
Integer id = resource.getId();
|
|
|
|
|
|
|
|
log.info("id: {} ", id);
|
|
|
|
|
|
|
|
if (id == null || id <= 0 ){
|
|
|
|
|
|
|
|
log.error("====== { ERROR , resource body is {} } =======" , resource.toString());
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
BuildRecordDto buildRecordDto = buildRecordService.findById(id);
|
|
|
|
|
|
|
|
// 通过是否含有 BuildRecordDto-params 来区分来源
|
|
|
|
// 通过是否含有 BuildRecordDto-params 来区分来源
|
|
|
|
List<Tag> collect;
|
|
|
|
List<Tag> collect;
|
|
|
|
|
|
|
|
|
|
|
|
List<String> remoteCollect;
|
|
|
|
List<String> remoteCollect;
|
|
|
|
|
|
|
|
|
|
|
|
// 获取需要进行去重的任务路径
|
|
|
|
// 获取需要进行去重的任务路径
|
|
|
|
String resultFilePath = getPreFilterPath(sendBigDataDTO);
|
|
|
|
String resultFilePath = getPreFilterPath(sendBigDataDTO);
|
|
|
|
|
|
|
|
|
|
|
|
if (StringUtils.isNotBlank(buildRecordDto.getParams())){
|
|
|
|
if (resource.getId() != null){
|
|
|
|
|
|
|
|
// 根据发送任务的Id来读取发送号码表
|
|
|
|
|
|
|
|
Integer id = resource.getId();
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
BuildRecordDto buildRecordDto = buildRecordService.findById(id);
|
|
|
|
// 本地表库中的记录的源
|
|
|
|
// 本地表库中的记录的源
|
|
|
|
collect = getTagsFromLocalDBRec(resource, sendBigDataDTO, buildRecordDto);
|
|
|
|
collect = getTagsFromLocalDBRec(resource, sendBigDataDTO, buildRecordDto);
|
|
|
|
// 进行去重逻辑
|
|
|
|
// 进行去重逻辑
|
|
|
@ -126,13 +122,16 @@ public class SendBigDataTask {
|
|
|
|
// 对需要发送的字段进行发送
|
|
|
|
// 对需要发送的字段进行发送
|
|
|
|
AtomicLong atomicLong = batchSend(collect, sendBigDataDTO, buildRecordDto);
|
|
|
|
AtomicLong atomicLong = batchSend(collect, sendBigDataDTO, buildRecordDto);
|
|
|
|
finishSendThenUpdateRec(resource, buildRecordDto, atomicLong);
|
|
|
|
finishSendThenUpdateRec(resource, buildRecordDto, atomicLong);
|
|
|
|
}else {
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if (remoteRecord.getId() != null){
|
|
|
|
|
|
|
|
RemoteRecordDto remoteRecordDto = remoteRecordService.findById(remoteRecord.getId());
|
|
|
|
// 远程文件存储形式的源
|
|
|
|
// 远程文件存储形式的源
|
|
|
|
remoteCollect = getRemoteRecFromLocal(buildRecordDto);
|
|
|
|
remoteCollect = getRemoteRecFromLocal(remoteRecordDto);
|
|
|
|
// 进行去重逻辑
|
|
|
|
// 进行去重逻辑
|
|
|
|
remoteCollect = getNotDuplicateRemoteContent(remoteCollect, resultFilePath);
|
|
|
|
remoteCollect = getNotDuplicateRemoteContent(remoteCollect, resultFilePath);
|
|
|
|
// 批量发送信息并且更新记录
|
|
|
|
// 批量发送信息并且更新记录
|
|
|
|
AtomicLong atomicLong = batchSendRemote(remoteCollect, sendBigDataDTO, buildRecordDto);
|
|
|
|
AtomicLong atomicLong = batchSendRemote(remoteCollect, sendBigDataDTO, resource);
|
|
|
|
finishSendThenUpdateRemoteRec(remoteRecord, atomicLong);
|
|
|
|
finishSendThenUpdateRemoteRec(remoteRecord, atomicLong);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
@ -174,7 +173,7 @@ public class SendBigDataTask {
|
|
|
|
RemoteRecord remote = new RemoteRecord();
|
|
|
|
RemoteRecord remote = new RemoteRecord();
|
|
|
|
BeanUtil.copyProperties(remoteRecord, remote);
|
|
|
|
BeanUtil.copyProperties(remoteRecord, remote);
|
|
|
|
remote.setTag(SUCCESS_TAG);
|
|
|
|
remote.setTag(SUCCESS_TAG);
|
|
|
|
remote.setWeight(10);
|
|
|
|
remote.setWeight(10); // 成功发送后把成功发送的记录权重值设置为 10
|
|
|
|
remote.setSuccessSendCount(atomicLong.get());
|
|
|
|
remote.setSuccessSendCount(atomicLong.get());
|
|
|
|
remoteRecordService.update(remote);
|
|
|
|
remoteRecordService.update(remote);
|
|
|
|
}
|
|
|
|
}
|
|
|
@ -239,8 +238,8 @@ public class SendBigDataTask {
|
|
|
|
return resultFilePath;
|
|
|
|
return resultFilePath;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
private List<String> getRemoteRecFromLocal(BuildRecordDto buildRecordDto) {
|
|
|
|
private List<String> getRemoteRecFromLocal(RemoteRecordDto remoteRecordDto) {
|
|
|
|
String localFilePath = buildRecordDto.getLocalFilePath();
|
|
|
|
String localFilePath = remoteRecordDto.getLocalSavePath();
|
|
|
|
if (StringUtils.isBlank(localFilePath)){
|
|
|
|
if (StringUtils.isBlank(localFilePath)){
|
|
|
|
log.error("SendBigDataTask|getRemoteRecFromLocal , localFilePath is null!");
|
|
|
|
log.error("SendBigDataTask|getRemoteRecFromLocal , localFilePath is null!");
|
|
|
|
}
|
|
|
|
}
|
|
|
@ -322,14 +321,14 @@ public class SendBigDataTask {
|
|
|
|
return successCount;
|
|
|
|
return successCount;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
private AtomicLong batchSendRemote(List<String> collect, SendBigDataDTO sendRecordDTO, BuildRecordDto buildRecordDto) {
|
|
|
|
private AtomicLong batchSendRemote(List<String> collect, SendBigDataDTO sendRecordDTO, BuildRecord resource) {
|
|
|
|
AtomicLong successCount = new AtomicLong(0L);
|
|
|
|
AtomicLong successCount = new AtomicLong(0L);
|
|
|
|
List<List<String>> partition = Lists.partition(collect, SEND_LIMIT);
|
|
|
|
List<List<String>> partition = Lists.partition(collect, SEND_LIMIT);
|
|
|
|
partition.forEach(
|
|
|
|
partition.forEach(
|
|
|
|
list->{
|
|
|
|
list->{
|
|
|
|
// 调用推送地址进行推送
|
|
|
|
// 调用推送地址进行推送
|
|
|
|
PushDBJsonContent pushDBJsonContent = new PushDBJsonContent();
|
|
|
|
PushDBJsonContent pushDBJsonContent = new PushDBJsonContent();
|
|
|
|
pushDBJsonContent.setActId(buildRecordDto.getTaskBuildId());
|
|
|
|
pushDBJsonContent.setActId(resource.getTaskBuildId());
|
|
|
|
pushDBJsonContent.setActName(sendRecordDTO.getSendName());
|
|
|
|
pushDBJsonContent.setActName(sendRecordDTO.getSendName());
|
|
|
|
// 加入每一个号码对应接口字段
|
|
|
|
// 加入每一个号码对应接口字段
|
|
|
|
List<PushDBJsonContent.Client> clientList = new ArrayList<>();
|
|
|
|
List<PushDBJsonContent.Client> clientList = new ArrayList<>();
|
|
|
|