package com.highdatas.mdm.service.impl;
|
|
import com.highdatas.mdm.entity.*;
|
import com.highdatas.mdm.pojo.CodeMsg;
|
import com.highdatas.mdm.pojo.Page;
|
import com.highdatas.mdm.pojo.Result;
|
import com.highdatas.mdm.service.*;
|
import com.highdatas.mdm.util.Constant;
|
import com.highdatas.mdm.util.DbUtils;
|
import com.highdatas.mdm.util.pool.MqEntity;
|
import com.highdatas.mdm.util.pool.MqMessage;
|
import com.highdatas.mdm.util.pool.PassiveMqThreadPoolExecutor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang3.StringUtils;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.boot.context.properties.ConfigurationProperties;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.PostConstruct;
|
import java.util.ArrayList;
|
import java.util.Date;
|
import java.util.List;
|
import java.util.concurrent.*;
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
/**
|
* @author kimi
|
* @description
|
* @date 2020-04-14 12:58
|
*/
|
|
@Service
|
@Slf4j
|
@ConfigurationProperties(prefix = "pool")
|
public class DispenseServiceImpl implements DispenseService {
|
@Autowired
|
MasterDataService masterDataService;
|
@Autowired
|
IMasterAuthorService masterAuthorService;
|
@Autowired
|
ISysViewService viewService;
|
@Autowired
|
IMaintainService maintainService;
|
@Autowired
|
IMenuMappingService menuMappingService;
|
@Autowired
|
ISysDispenseLogsService dispenseLogsService;
|
|
@Value("${pool.coresize}")
|
private String coreSizeStr;
|
|
@Value("${pool.aes}")
|
private Boolean aes;
|
@Value("${pool.aesurl}")
|
private String aesUrl;
|
|
private int coreSize;
|
private Integer queueSize = 5;
|
private ExecutorService priorityExecutorService;
|
private volatile PriorityBlockingQueue<Runnable> priorityQueue;
|
private volatile LinkedBlockingQueue infiniteQueue;
|
private volatile CopyOnWriteArrayList<MqMessage> passiveRequestList;
|
private ThreadPoolExecutor infiniteExecutorService;
|
|
/**
|
*
|
* @description: 初始化的时候创建主动，被动两个排队队列
|
* @return: void
|
*
|
*/
|
|
@PostConstruct
|
public void init() {
|
|
if (StringUtils.isEmpty(coreSizeStr)) {
|
//IO密集型 设置默认 cpu数量 预留 主动和被动两组线程池
|
coreSize = Double.valueOf(Runtime.getRuntime().availableProcessors()).intValue();
|
} else {
|
coreSize = Integer.valueOf(coreSizeStr);
|
}
|
log.info("Queue_Consume_thread_size:{}",coreSize);
|
this.priorityQueue = new PriorityBlockingQueue<>(queueSize);
|
this.priorityExecutorService = new PassiveMqThreadPoolExecutor(coreSize, coreSize, 0L, TimeUnit.MILLISECONDS, priorityQueue);
|
this.infiniteQueue = new LinkedBlockingQueue();
|
this.infiniteExecutorService = new ThreadPoolExecutor(coreSize, coreSize, 0L, TimeUnit.MILLISECONDS, infiniteQueue,new ThreadPoolExecutor.AbortPolicy());
|
this.passiveRequestList = new CopyOnWriteArrayList<>();
|
createPassiveListenerThread();
|
}
|
/**
|
*
|
* @description: 获取是否需要
|
* @return: 是否需要aes加密
|
*
|
*/
|
@Override
|
public Boolean getAes() {
|
return aes;
|
}
|
/**
|
*
|
* @description: 获取ars加密请求的路径
|
* @return: 加密接口路径
|
*
|
*/
|
@Override
|
public String getAesUrl() {
|
return aesUrl;
|
}
|
|
/**
|
*
|
* @description: 创建主动分发队列
|
* @return: void
|
*
|
*/
|
|
private void createPassiveListenerThread() {
|
new Thread(() -> {
|
log.info("被动接口准备待命。");
|
while (true) {
|
try{
|
if (passiveRequestList.size() == 0) {
|
Thread.sleep(5 * 1000);
|
}
|
List<MqMessage> removedList = new ArrayList<>();
|
for (MqMessage mqMessage : passiveRequestList) {
|
MqEntity mqEntity = mqMessage.getMqEntity();
|
AtomicInteger pageNo = mqEntity.getPageNo();
|
if (pageNo != null && pageNo.get() == mqEntity.getPages().get()) {
|
removedList.add(mqMessage);
|
} else {
|
MqMessage nextSubMq = createNextSubMq(mqEntity);
|
infiniteExecutorService.execute(nextSubMq);
|
}
|
|
}
|
for (MqMessage mqMessage : removedList) {
|
passiveRequestList.remove(mqMessage);
|
}
|
|
}
|
catch (Exception e){
|
e.printStackTrace();
|
//log
|
}
|
}
|
}).start();
|
}
|
|
/**
|
*
|
* @description: 创建下页分发的数据包信息
|
* @param mqEntity 当前分发的entity数据
|
* @return: void
|
*
|
*/
|
private MqMessage createNextSubMq(MqEntity mqEntity) {
|
String type = mqEntity.getType();
|
AtomicInteger pageNoAI = mqEntity.getPageNo();
|
String logId = mqEntity.getLogId();
|
SysDispenseLogs logs = dispenseLogsService.selectById(logId);
|
|
if(pageNoAI == null) {
|
String userId = mqEntity.getUserId();
|
TUser user = DbUtils.getUserById(userId);
|
if (user == null && logs != null) {
|
logs.setErrorInfo(CodeMsg.USER_NOT_MATHED.getMsg()).updateById();
|
} else if (logs != null) {
|
logs.setUserName(user.getUserName());
|
logs.updateById();
|
}
|
if (type.equalsIgnoreCase(Constant.Master)) {
|
String dataId = mqEntity.getDataId();
|
String tableNameByMenu = menuMappingService.getTableNameByMenu(dataId);
|
Maintain maintain = maintainService.selectById(mqEntity.getMaintainId());
|
MasterAuthor masterAuthor = new MasterAuthor().setTableName(tableNameByMenu);
|
masterAuthor.setIncrement(mqEntity.getIncrement());
|
Page initPageInfo = masterAuthorService.getInitPageInfo(masterAuthor, maintain, user, mqEntity.getIncrement());
|
AtomicInteger pagesAI = new AtomicInteger(initPageInfo.getPages());
|
mqEntity.setPages(pagesAI);
|
mqEntity.setTotalSize(new AtomicInteger(Long.valueOf(initPageInfo.getRecordCount()).intValue()));
|
AtomicInteger pageSizeAI = new AtomicInteger(initPageInfo.getPageSize());
|
mqEntity.setPageSize(pageSizeAI);
|
if (logs != null) {
|
SysMenu menuByTableName = menuMappingService.getMenuByTableName(maintain.getTableName());
|
if (menuByTableName != null) {
|
logs.setName(menuByTableName.getName());
|
}
|
logs.setVersion(maintain.getVersion());
|
logs.updateById();
|
}
|
|
} else {
|
Page initPageInfo = viewService.getInitPageInfo(mqEntity.getDataId());
|
AtomicInteger pagesAI = new AtomicInteger(initPageInfo.getPages());
|
mqEntity.setPages(pagesAI);
|
AtomicInteger pageSizeAI = new AtomicInteger(initPageInfo.getPageSize());
|
mqEntity.setPageSize(pageSizeAI);
|
mqEntity.setTotalSize(new AtomicInteger(Long.valueOf(initPageInfo.getRecordCount()).intValue()));
|
if (logs != null) {
|
SysView sysView = viewService.selectById(mqEntity.getDataId());
|
if (sysView != null) {
|
logs.setName(sysView.getName());
|
logs.updateById();
|
}
|
}
|
}
|
pageNoAI = new AtomicInteger();
|
mqEntity.setPageNo(pageNoAI);
|
}
|
int i = pageNoAI.addAndGet(1);
|
MqMessage mqMessage = new MqMessage(mqEntity);
|
return mqMessage;
|
}
|
/**
|
*
|
* @description: 添加到主动分发队列
|
* @param message 数据包信息
|
* @return: 是否添加成功
|
*
|
*/
|
@Override
|
public boolean pushActiveMq(MqMessage message) {
|
return pushActiveMq(message, null);
|
}
|
@Override
|
public boolean pushActiveMq(MqMessage message, String touchType) {
|
MqEntity mqEntity = message.getMqEntity();
|
SysDispenseLogs preUnSuccessByMqEntity = dispenseLogsService.getPreUnSuccessByMqEntity(mqEntity);
|
if (preUnSuccessByMqEntity == null) {
|
SysDispenseLogs logByMqEntity = dispenseLogsService.createLogByMqEntity(mqEntity);
|
if (StringUtils.isEmpty(touchType)) {
|
logByMqEntity.setTouchType(touchType);
|
} else {
|
logByMqEntity.setTouchType("sys");
|
}
|
|
|
logByMqEntity.updateById();
|
|
mqEntity.setLogId(logByMqEntity.getId());
|
|
} else {
|
mqEntity.setLogId(preUnSuccessByMqEntity.getId());
|
mqEntity.setMsgKey(preUnSuccessByMqEntity.getKeyId());
|
mqEntity.setPageNo(new AtomicInteger(preUnSuccessByMqEntity.getPageNo()));
|
mqEntity.setPageSize(new AtomicInteger(preUnSuccessByMqEntity.getPageSize()));
|
mqEntity.setPages(new AtomicInteger(preUnSuccessByMqEntity.getPages()));
|
}
|
|
this.passiveRequestList.add(message);
|
|
return true;
|
}
|
/**
|
*
|
* @description: 添加到被动分发队列
|
* @param message 数据包信息
|
* @param touchType 触发类型
|
* @param logs 日志对象
|
* @return: 是否添加成功
|
*
|
*/
|
@Override
|
public Result pushPassiveMq(MqMessage message, String touchType, SysDispenseLogs logs) {
|
try{
|
MqMessage preMsg = null;
|
if (queueSize.equals(priorityQueue.size())) {
|
logs.setResult(CodeMsg.ADDQUEUE_OVER.getMsg()).updateById();
|
return Result.error(CodeMsg.ADDQUEUE_OVER);
|
}
|
MqEntity mqEntity = message.getMqEntity();
|
for (Runnable runnable : priorityQueue) {
|
if (!(runnable instanceof MqMessage)) {
|
continue;
|
}
|
MqMessage mqMessage = (MqMessage) runnable;
|
String code = mqMessage.getMqEntity().getMsgCode();
|
if (StringUtils.isEmpty(code)){
|
continue;
|
}
|
if (code.equalsIgnoreCase(mqEntity.getMsgCode())) {
|
preMsg = (MqMessage) runnable;
|
break;
|
}
|
}
|
if (preMsg != null) {
|
int cnt = preMsg.getCnt();
|
preMsg.setCnt(cnt + 1);
|
logs.setResult(CodeMsg.REPEAT_ERROR.getMsg()).updateById();
|
return Result.error(CodeMsg.REPEAT_ERROR);
|
|
}else {
|
message.setTime(new Date());
|
message.setCnt(0);
|
}
|
priorityExecutorService.execute(message);
|
SysDispenseLogs preUnSuccessByMqEntity = dispenseLogsService.getPreUnSuccessByMqEntity(mqEntity);
|
if (preUnSuccessByMqEntity == null) {
|
logs.setResult(CodeMsg.ADDQUEUE_SUCCESS.getMsg()).updateById();
|
} else {
|
mqEntity.setMsgKey(preUnSuccessByMqEntity.getKeyId());
|
mqEntity.setPageNo(new AtomicInteger(preUnSuccessByMqEntity.getPageNo()));
|
mqEntity.setPageSize(new AtomicInteger(preUnSuccessByMqEntity.getPageSize()));
|
mqEntity.setPages(new AtomicInteger(preUnSuccessByMqEntity.getPages()));
|
}
|
|
return Result.success(CodeMsg.ADDQUEUE_SUCCESS);
|
}catch (Exception e) {
|
log.error(e.getMessage());
|
logs.setErrorInfo(e.getMessage());
|
return Result.success(CodeMsg.ADDQUEUE_FAIL);
|
}
|
}
|
|
/**
|
*
|
* @description: 获取主动分发队列中的排队数
|
* @return: 主动分发排队数
|
*
|
*/
|
|
|
@Override
|
public Integer passiveQueueSize() {
|
return priorityQueue.size();
|
}
|
/**
|
*
|
* @description: 获取被动分发队列中的排队数
|
* @return: 被动分发排队数
|
*
|
*/
|
@Override
|
public Integer avtiveQueueSize() {
|
return infiniteQueue.size();
|
}
|
|
|
}
|
