package com.highdatas.mdm.service.impl;
|
|
import com.highdatas.mdm.entity.Maintain;
|
import com.highdatas.mdm.service.DispenseService;
|
import com.highdatas.mdm.service.MasterDataService;
|
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.stereotype.Service;
|
|
import javax.annotation.PostConstruct;
|
import java.util.Date;
|
import java.util.concurrent.*;
|
|
/**
|
* @author kimi
|
* @description
|
* @date 2020-04-14 12:58
|
*/
|
|
@Service
|
@Slf4j
|
public class DispenseServiceImpl implements DispenseService {
|
@Autowired
|
MasterDataService masterDataService;
|
@Value("${pool.coresize}")
|
private Integer coreSize;
|
private ExecutorService executorService;
|
private PriorityBlockingQueue<Runnable> queue;
|
private LinkedBlockingQueue activeQueue;
|
private ThreadPoolExecutor activeExecutorService;
|
|
@PostConstruct
|
public void init() {
|
if (coreSize == null && coreSize == 0) {
|
//IO密集型 设置默认 cpu数量 预留 主动和被动两组线程池
|
coreSize = Double.valueOf(Runtime.getRuntime().availableProcessors()).intValue();
|
}
|
log.info("Queue_Consume_thread_size:{}",coreSize);
|
this.queue = new PriorityBlockingQueue<>();
|
this.executorService = new PassiveMqThreadPoolExecutor(coreSize, coreSize, 0L, TimeUnit.MILLISECONDS, queue);
|
this.activeQueue = new LinkedBlockingQueue();
|
this.activeExecutorService = new ThreadPoolExecutor(coreSize, coreSize, 0L, TimeUnit.MILLISECONDS, activeQueue,new ThreadPoolExecutor.AbortPolicy());
|
}
|
|
@Override
|
public boolean pushActiveMq(MqMessage message) {
|
try{
|
activeExecutorService.execute(message);
|
return true;
|
}catch (Exception e) {
|
log.error(e.getMessage());
|
return false;
|
}
|
}
|
|
@Override
|
public boolean pushPassiveMq(MqMessage message) {
|
try{
|
MqMessage preMsg = null;
|
for (Runnable runnable : queue) {
|
if (!(runnable instanceof MqMessage)) {
|
continue;
|
}
|
MqMessage mqMessage = (MqMessage) runnable;
|
String code = mqMessage.getCode();
|
if (StringUtils.isEmpty(code)){
|
continue;
|
}
|
if (code.equalsIgnoreCase(message.getCode())) {
|
preMsg = (MqMessage) runnable;
|
}
|
}
|
if (preMsg != null) {
|
int cnt = preMsg.getCnt();
|
preMsg.setCnt(cnt + 1);
|
message.printRepeat();
|
}else {
|
message.setTime(new Date());
|
message.setCnt(0);
|
}
|
executorService.execute(message);
|
return true;
|
}catch (Exception e) {
|
log.error(e.getMessage());
|
return false;
|
}
|
}
|
|
@Override
|
public PriorityBlockingQueue<Runnable> getQueue() {
|
return queue;
|
}
|
|
@Override
|
public Integer passiveQueueSize() {
|
return queue.size();
|
}
|
|
@Override
|
public Integer avtiveQueueSize() {
|
return activeQueue.size();
|
}
|
|
private void createView(Maintain baseMaintain) {
|
String maintainTableName = baseMaintain.getTableName();
|
String version = baseMaintain.getVersion();
|
|
}
|
}
|
