package com.dingzhuo.compute.engine.actor.indexcalc;
|
|
import static org.apache.commons.lang3.StringUtils.equalsIgnoreCase;
|
|
import akka.actor.ActorSelection;
|
import akka.actor.ActorSystem;
|
import akka.actor.Cancellable;
|
import akka.actor.UntypedAbstractActor;
|
import akka.event.Logging;
|
import akka.event.LoggingAdapter;
|
import com.dingzhuo.compute.engine.message.ExecuteType;
|
import com.dingzhuo.compute.engine.message.calculation.CalculateMessage;
|
import com.dingzhuo.compute.engine.message.save.SaveMessage;
|
import com.dingzhuo.compute.engine.utils.ActorUtil;
|
import com.dingzhuo.compute.engine.utils.ServiceProvicer;
|
import com.dingzhuo.energy.common.utils.time.TimeManager;
|
import com.dingzhuo.energy.dataservice.domain.DataItem;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.TimeUnit;
|
import org.joda.time.DateTime;
|
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
import org.springframework.context.annotation.Scope;
|
import org.springframework.stereotype.Component;
|
import scala.concurrent.duration.Duration;
|
import scala.concurrent.duration.FiniteDuration;
|
|
@Component("savePeriodActor")
|
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
public class SavePeriodActor extends UntypedAbstractActor {
|
|
public static final String ACTOR_NAME = "savePeriodActor";
|
LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
|
private final ActorSystem actorSystem;
|
private Cancellable timer;
|
private ActorSelection calculateActor;
|
private ConcurrentHashMap<String, SaveMessage> cacheData = new ConcurrentHashMap<>();
|
|
public SavePeriodActor(ActorSystem actorSystem) {
|
this.actorSystem = actorSystem;
|
}
|
|
@Override
|
public void preStart() throws Exception {
|
super.preStart();
|
calculateActor = getContext()
|
.actorSelection(ActorUtil.getActorAddress(CalculationIndexActor.ACTOR_NAME));
|
FiniteDuration interval = Duration.create(10, TimeUnit.SECONDS);
|
FiniteDuration delay = Duration.Zero();
|
timer = actorSystem.scheduler()
|
.scheduleAtFixedRate(delay, interval, self(), Message.TIMER, actorSystem.dispatcher(),
|
self());
|
}
|
|
@Override
|
public void postStop() throws Exception {
|
super.postStop();
|
if (timer != null) {
|
timer.cancel();
|
}
|
}
|
|
@Override
|
public void onReceive(Object message) {
|
if (message instanceof Message) {
|
if (message == Message.TIMER) {
|
doSave();
|
}
|
} else if (message instanceof SaveMessage) {
|
SaveMessage saveMessage = (SaveMessage) message;
|
String key = String.format("%s:%s", saveMessage.getIndexId(), saveMessage.getTimeCode());
|
cacheData.put(key, saveMessage);
|
}
|
}
|
|
private void doSave() {
|
if (!cacheData.isEmpty()) {
|
List<SaveMessage> saveData = new ArrayList<>();
|
List<String> needRemoveKeys = new ArrayList<>();
|
List<CalculateMessage> postIndexMessage = new ArrayList<>();
|
cacheData.forEach((key, value) -> {
|
needRemoveKeys.add(key);
|
saveData.add(value);
|
value.getPostActorIds().forEach(actorId -> {
|
if (!equalsIgnoreCase(actorId,
|
ActorUtil.buildActorId(value.getIndexId(), value.getTimeType()))) {
|
CalculateMessage message =
|
new CalculateMessage(actorId, value.getTimeCode(),
|
value.getTimeType(),
|
ExecuteType.TIMER);
|
postIndexMessage.add(message);
|
}
|
});
|
});
|
|
for (String key : needRemoveKeys) {
|
cacheData.remove(key);
|
}
|
|
savePeriodData(saveData);
|
postIndexMessage.forEach(message -> calculateActor.tell(message, getSelf()));
|
}
|
}
|
|
private void savePeriodData(List<SaveMessage> saveData) {
|
List<DataItem> dataItems = new ArrayList<>();
|
saveData.forEach(data -> {
|
String timeCode = data.getTimeCode();
|
DataItem dataItem = new DataItem();
|
dataItem.setIndexId(data.getIndexId());
|
dataItem.setTimeCode(timeCode);
|
dataItem.setTimeType(data.getTimeType());
|
dataItem.setBeginTime(TimeManager.getBeginTime(timeCode));
|
dataItem.setEndTime(TimeManager.getEndTime(timeCode));
|
dataItem.setDataTime(TimeManager.getTime(timeCode));
|
dataItem.setValue(data.getValue());
|
dataItems.add(dataItem);
|
});
|
|
try {
|
ServiceProvicer.getPeriodDataService().save(dataItems);
|
} catch (Exception ex) {
|
log.error("批量保存失败！" + ex.getMessage());
|
dataItems.forEach(item -> {
|
try {
|
if (item != null) {
|
ServiceProvicer.getPeriodDataService().save(item);
|
}
|
} catch (Exception singleEx) {
|
log.error("单个保存失败！" + singleEx.getMessage());
|
}
|
});
|
}
|
|
log.error(DateTime.now().toString("yyyy-MM-dd HH:mm:ss.SSS"));
|
}
|
|
private enum Message {
|
/**
|
* 时间触发
|
*/
|
TIMER
|
}
|
}
|
