http://jira.mfd.msk:8088/browse/CLS-1 storage копипаст сервисов (осторожно-тудушки и конфига нет): готовы на 98%
This commit is contained in:
akurakin
parent
83463a1ebe
commit
1dee51257e
4 changed files with 356 additions and 0 deletions
|
|
@ -0,0 +1,216 @@
|
||||||
|
package ru.spcex.clearing.storage.services;
|
||||||
|
|
||||||
|
import com.hazelcast.aggregation.Aggregators;
|
||||||
|
import com.hazelcast.config.MapStoreConfig;
|
||||||
|
import com.hazelcast.core.HazelcastInstance;
|
||||||
|
import com.hazelcast.core.IAtomicLong;
|
||||||
|
import com.hazelcast.core.IMap;
|
||||||
|
import com.hazelcast.core.IdGenerator;
|
||||||
|
import com.hazelcast.projection.Projections;
|
||||||
|
import com.hazelcast.query.Predicates;
|
||||||
|
import org.apache.commons.lang3.StringUtils;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.beans.factory.DisposableBean;
|
||||||
|
import org.springframework.beans.factory.InitializingBean;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.dao.DataAccessException;
|
||||||
|
import org.springframework.jdbc.core.JdbcTemplate;
|
||||||
|
import ru.clearing.classes.objects.BusinessObject;
|
||||||
|
import ru.spcex.clearing.storage.base.DictionaryMapStore;
|
||||||
|
import ru.spcex.clearing.storage.base.SimpleObjectMapStore;
|
||||||
|
import ru.spcex.clearing.storage.utils.IMDGDistributedNames;
|
||||||
|
|
||||||
|
import java.time.DayOfWeek;
|
||||||
|
import java.time.LocalDate;
|
||||||
|
import java.time.ZoneId;
|
||||||
|
import java.util.*;
|
||||||
|
import java.util.concurrent.*;
|
||||||
|
import java.util.regex.Matcher;
|
||||||
|
import java.util.regex.Pattern;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
public abstract class AbstractHazelcastLifecycleSupport implements InitializingBean, DisposableBean {
|
||||||
|
/**
|
||||||
|
* Рабочая версия БД. Треьуется вручную сверять с DDL.sql и накручивать эту переменную.
|
||||||
|
*/
|
||||||
|
public abstract String getCheckDbVersion();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Обязательная сверка у этих классов SerialVersionUID во время подключения к Storage.
|
||||||
|
*/
|
||||||
|
public abstract Class[] getSerialVersionUIDClasses();
|
||||||
|
|
||||||
|
private final Logger log = LoggerFactory.getLogger(this.getClass());
|
||||||
|
|
||||||
|
private final HazelcastInstance hazelcastServerInstance;
|
||||||
|
private final JdbcTemplate jdbcTemplate;
|
||||||
|
// private final HazelcastClientListener clientListener;
|
||||||
|
// private final ITaskAdministrator startupTasksAdministrator;
|
||||||
|
|
||||||
|
public AbstractHazelcastLifecycleSupport(HazelcastInstance hazelcastServerInstance, JdbcTemplate jdbcTemplate) {
|
||||||
|
this.hazelcastServerInstance = hazelcastServerInstance;
|
||||||
|
this.jdbcTemplate = jdbcTemplate;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void afterPropertiesSet() {
|
||||||
|
// HazelcastCommon.loadClassesVersionsByStorage(hazelcastServerInstance, getSerialVersionUIDClasses());
|
||||||
|
// databaseVersionCheck();
|
||||||
|
|
||||||
|
long loadTime = System.currentTimeMillis();
|
||||||
|
|
||||||
|
try {
|
||||||
|
List<Callable<Long>> tasks = new ArrayList<>();
|
||||||
|
for (String mapName : hazelcastServerInstance.getConfig().getMapConfigs().keySet()) {
|
||||||
|
tasks.add(() -> {
|
||||||
|
Long maxKey = null;
|
||||||
|
MapStoreConfig mapStoreConfig = hazelcastServerInstance.getConfig().getMapConfig(mapName).getMapStoreConfig();
|
||||||
|
if (mapStoreConfig != null && mapStoreConfig.isEnabled()) {
|
||||||
|
long start = System.currentTimeMillis();
|
||||||
|
log.info("loading map {}", mapName);
|
||||||
|
IMap<Long, BusinessObject> map = hazelcastServerInstance.getMap(mapName);
|
||||||
|
map.loadAll(false);
|
||||||
|
int size = map.size();
|
||||||
|
long time = System.currentTimeMillis() - start;
|
||||||
|
log.info("{} {} rows loaded in {}ms", mapName, size, time);
|
||||||
|
|
||||||
|
Object mapStore = mapStoreConfig.getImplementation();
|
||||||
|
if (mapStore instanceof SimpleObjectMapStore) {
|
||||||
|
String tableName = ((SimpleObjectMapStore) mapStore).getTableName();
|
||||||
|
maxKey = jdbcTemplate.queryForObject("select max(id) from " + tableName, Long.class);
|
||||||
|
} else if (mapStore instanceof DictionaryMapStore) {
|
||||||
|
// для Dictionary не используется общий id генератор
|
||||||
|
// } else if (mapStore instanceof FrontendUserSessionMapStore) {
|
||||||
|
// // не используется общий id генератор
|
||||||
|
} else {
|
||||||
|
throw new RuntimeException("unknown map store implementation " + mapStore);
|
||||||
|
}
|
||||||
|
log.debug("{} max(id)={}", mapName, maxKey);
|
||||||
|
}
|
||||||
|
|
||||||
|
return maxKey;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
int threadCount = Runtime.getRuntime().availableProcessors();// todo config * Config.get().getRoot().getSettings().getInitHazelcastThreadMultiplier();
|
||||||
|
log.info("Initializing threads count = {}", threadCount);
|
||||||
|
ExecutorService executor = Executors.newWorkStealingPool(threadCount);
|
||||||
|
List<Future<Long>> results = executor.invokeAll(tasks);
|
||||||
|
long maxKey = 0L;
|
||||||
|
for (Future<Long> result : results) {
|
||||||
|
Long maxKeyResult = result.get();
|
||||||
|
if (maxKeyResult != null) {
|
||||||
|
maxKey = Math.max(maxKey, maxKeyResult);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
IdGenerator generator = hazelcastServerInstance.getIdGenerator(IMDGDistributedNames.MAP_SEQUENCE_NAME);
|
||||||
|
boolean generatorResult = generator.init(maxKey);
|
||||||
|
if (generatorResult) {
|
||||||
|
log.info("IDGenerator {} success init by {}", IMDGDistributedNames.MAP_SEQUENCE_NAME, maxKey);
|
||||||
|
// makeSchedulerAllTodayMap();
|
||||||
|
} else {
|
||||||
|
log.info("IDGenerator {} already initialized in other node", IMDGDistributedNames.MAP_SEQUENCE_NAME);
|
||||||
|
}
|
||||||
|
|
||||||
|
// MapStoreConfig mapStoreConfig = hazelcastServerInstance.getConfig().getMapConfig(HazelcastDistributedNames.Map_FixedIncomeProductExecution).getMapStoreConfig();
|
||||||
|
// FixedIncomeProductExecutionMapStore fixedIncomeProductExecutionMapStore = (FixedIncomeProductExecutionMapStore) mapStoreConfig.getImplementation();
|
||||||
|
// Long maxExecutionNumber = jdbcTemplate.queryForObject("select max(executionnumber) from " + fixedIncomeProductExecutionMapStore.getTableName(), Long.class);
|
||||||
|
// log.info("{} max(executionnumber)={}", fixedIncomeProductExecutionMapStore.getTableName(), maxExecutionNumber);
|
||||||
|
// if (maxExecutionNumber != null) {
|
||||||
|
// if (Integer.MAX_VALUE - maxExecutionNumber < 1000000)
|
||||||
|
// log.error("max(executionnumber) close to int32 max value");
|
||||||
|
// IAtomicLong executionNumberAtomicLong = hazelcastServerInstance.getAtomicLong(HazelcastDistributedNames.AtomicLong_ExecutionNumber);
|
||||||
|
// executionNumberAtomicLong.set(maxExecutionNumber);
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
// if (startupTasksAdministrator != null) {
|
||||||
|
// startupTasksAdministrator.createTasks();
|
||||||
|
// }
|
||||||
|
} catch (InterruptedException | ExecutionException e) {
|
||||||
|
throw new RuntimeException("MapStore multithreaded not complete.", e);
|
||||||
|
}
|
||||||
|
|
||||||
|
// HazelcastCommon.otcSystem_setStorageState(true, hazelcastServerInstance); todo в других модулях может быть проверка на это, и должна быть.
|
||||||
|
// hazelcastServerInstance.getClientService().addClientListener(clientListener);
|
||||||
|
loadTime = System.currentTimeMillis() - loadTime;
|
||||||
|
log.info("All map load time {} ms", loadTime);
|
||||||
|
|
||||||
|
// TextErrorService.setHazelcast(hazelcastServerInstance);
|
||||||
|
|
||||||
|
// clusterStatistic();
|
||||||
|
}
|
||||||
|
|
||||||
|
// /**
|
||||||
|
// * Подсчитывает кол-во запусков, первый запуск. Пишет в лог.
|
||||||
|
// * Storage note status: ...
|
||||||
|
// */
|
||||||
|
// protected void clusterStatistic() {
|
||||||
|
// try {
|
||||||
|
// boolean isFirstNodeReady = HazelcastCommon.otcSystem_setStorageInfo(new Date(), true, hazelcastServerInstance);
|
||||||
|
// String msg = "Storage node status: " + (isFirstNodeReady ? "first Storage node" : "second node")
|
||||||
|
// + ", first node start at " + HazelcastCommon.otcSystem_getFirstStorageTime(hazelcastServerInstance);
|
||||||
|
// String firstOtcVersion = HazelcastCommon.otcSystem_getFirstStorageVersion(hazelcastServerInstance);
|
||||||
|
// if (firstOtcVersion != null)
|
||||||
|
// msg += "(OTC " + firstOtcVersion + ")";
|
||||||
|
// msg += ", count of all storage connection " + HazelcastCommon.otcSystem_getStorageConnectCount(hazelcastServerInstance) + ".";
|
||||||
|
// msg += " Hazelcast cluster members: " + hazelcastServerInstance.getCluster().getMembers().size() + ".";
|
||||||
|
// log.info(msg);
|
||||||
|
// } catch (Exception e) {
|
||||||
|
// log.warn("Error at print Storage cluster info.", e);
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void destroy() {
|
||||||
|
hazelcastServerInstance.shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
|
protected void databaseVersionCheck() throws RuntimeException, IllegalArgumentException {
|
||||||
|
log.debug("Validate DB version, expected \"{}\"", getCheckDbVersion());
|
||||||
|
String dbVersion = null;
|
||||||
|
try {
|
||||||
|
dbVersion = jdbcTemplate.queryForObject("select version from VERSIONEDID", String.class);
|
||||||
|
log.info("DB version is \"{}\"", dbVersion);
|
||||||
|
if (StringUtils.isBlank(dbVersion)) {
|
||||||
|
String msg = "Database not contain version information (table VERSIONEDID)";//TextErrorService.text(StorageErrors.StorageErrors_WrongDBVersion) + " Database not contain version information (table VERSIONEDID)";
|
||||||
|
log.error(msg);
|
||||||
|
throw new RuntimeException(msg);
|
||||||
|
}
|
||||||
|
if (!isAllowDBVersion(getCheckDbVersion(), dbVersion)) {
|
||||||
|
String msg = " platform " + getCheckDbVersion() + ", db " + dbVersion;//TextErrorService.text(StorageErrors.StorageErrors_WrongDBVersion) + " platform " + getCheckDbVersion() + ", db " + dbVersion;
|
||||||
|
log.error(msg);
|
||||||
|
throw new RuntimeException(msg);
|
||||||
|
}
|
||||||
|
} catch (DataAccessException | IllegalArgumentException e) { // EmptyResultDataAccessException
|
||||||
|
String msg = e.toString();//TextErrorService.text(StorageErrors.StorageErrors_WrongDBVersion) + " platform " + getCheckDbVersion() + ", db " + dbVersion;
|
||||||
|
log.error(msg);
|
||||||
|
throw new RuntimeException(msg, e); // Ошибка 10002 - Версия Бд не поддерживается
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Првоерка версии. Формат: 1.2.*
|
||||||
|
*
|
||||||
|
* @param moduleVersion версия кода
|
||||||
|
* @param dbVersion версия базы
|
||||||
|
* @return true - версии правильные, false - версия БД устарела или не совпадает.
|
||||||
|
* @throws IllegalArgumentException если неправильно распарсилась версия, в частности может случиться NumberFormatException.
|
||||||
|
*/
|
||||||
|
static boolean isAllowDBVersion(String moduleVersion, String dbVersion) throws IllegalArgumentException {
|
||||||
|
Pattern versionPattern = Pattern.compile("(\\d+)\\.(\\d+)(\\..*){0,1}"); // (\d+)\.(\d+)(\..*){0,1}
|
||||||
|
Matcher mModule = versionPattern.matcher(moduleVersion);
|
||||||
|
Matcher mDB = versionPattern.matcher(dbVersion);
|
||||||
|
if (!mModule.find())
|
||||||
|
throw new IllegalArgumentException("Module version have wrong format: " + moduleVersion);
|
||||||
|
if (!mDB.find())
|
||||||
|
throw new IllegalArgumentException("Version from database have wrong format: " + moduleVersion);
|
||||||
|
int majorModule = Integer.parseInt(mModule.group(1));
|
||||||
|
int minorModule = Integer.parseInt(mModule.group(2));
|
||||||
|
int majorDB = Integer.parseInt(mDB.group(1));
|
||||||
|
int minorDB = Integer.parseInt(mDB.group(2));
|
||||||
|
return majorDB == majorModule && minorDB >= minorModule;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,62 @@
|
||||||
|
package ru.spcex.clearing.storage.services;
|
||||||
|
|
||||||
|
import com.hazelcast.core.EntryEvent;
|
||||||
|
import com.hazelcast.core.HazelcastInstance;
|
||||||
|
import com.hazelcast.map.listener.EntryAddedListener;
|
||||||
|
import com.hazelcast.map.listener.EntryRemovedListener;
|
||||||
|
import com.hazelcast.map.listener.EntryUpdatedListener;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.beans.factory.InitializingBean;
|
||||||
|
import ru.clearing.classes.objects.BusinessEvent;
|
||||||
|
import ru.spcex.clearing.storage.utils.IMDGDistributedNames;
|
||||||
|
|
||||||
|
import java.time.Instant;
|
||||||
|
|
||||||
|
public abstract class AbstractUpdateMapService implements InitializingBean, EntryAddedListener<Long, Object>, EntryUpdatedListener<Long, Object>, EntryRemovedListener<Long, Object> {
|
||||||
|
private final Logger log = LoggerFactory.getLogger(this.getClass());
|
||||||
|
|
||||||
|
protected final String EVENT_CREATE="CREATE";
|
||||||
|
protected final String EVENT_UPDATE="UPDATE";
|
||||||
|
protected final String EVENT_DELETE="DELETE";
|
||||||
|
|
||||||
|
protected final HazelcastInstance hazelcastServerInstance;
|
||||||
|
|
||||||
|
public AbstractUpdateMapService(HazelcastInstance hazelcastServerInstance) {
|
||||||
|
this.hazelcastServerInstance = hazelcastServerInstance;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void afterPropertiesSet() {
|
||||||
|
// if (!HazelcastCommon.otcSystem_getStorageState(hazelcastServerInstance))
|
||||||
|
// log.warn("UpdateMapService started before Storage had ready!");
|
||||||
|
addingListenersToCards();
|
||||||
|
log.debug("UpdateMapService started.");
|
||||||
|
}
|
||||||
|
|
||||||
|
protected abstract void addingListenersToCards();
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void entryAdded(EntryEvent<Long, Object> event) {
|
||||||
|
entryModified(event, EVENT_CREATE);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void entryUpdated(EntryEvent<Long, Object> event) {
|
||||||
|
entryModified(event, EVENT_UPDATE);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void entryRemoved(EntryEvent<Long, Object> event) {
|
||||||
|
entryModified(event, EVENT_DELETE);
|
||||||
|
}
|
||||||
|
|
||||||
|
protected abstract void entryModified(EntryEvent<Long, Object> event, String eventType);
|
||||||
|
|
||||||
|
protected void createBusinessEvent(BusinessEvent businessEvent, String eventType) {
|
||||||
|
businessEvent.setId(hazelcastServerInstance.getIdGenerator(IMDGDistributedNames.MAP_SEQUENCE_NAME).newId());
|
||||||
|
businessEvent.setEventTime(Instant.now());
|
||||||
|
businessEvent.setUserId(0L); // todo откуда брать пользователя, причину изменений?
|
||||||
|
log.debug("business event created {} {}", businessEvent.getClass().getName(), businessEvent);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,37 @@
|
||||||
|
package ru.spcex.clearing.storage.services;
|
||||||
|
|
||||||
|
import com.hazelcast.core.HazelcastInstance;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.jdbc.core.JdbcTemplate;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
import ru.clearing.classes.ConstSerializable;
|
||||||
|
import ru.clearing.platform.dictionary.ConstDictionarySerializable;
|
||||||
|
|
||||||
|
@Service
|
||||||
|
public class HazelcastLifecycleSupport extends AbstractHazelcastLifecycleSupport {
|
||||||
|
/**
|
||||||
|
* Рабочая версия БД. Треьуется вручную сверять с DDL.sql и накручивать эту переменную.
|
||||||
|
*/
|
||||||
|
public static final String CHECK_DB_VERSION = "2.33";
|
||||||
|
|
||||||
|
private final Logger log = LoggerFactory.getLogger(this.getClass());
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public HazelcastLifecycleSupport(HazelcastInstance hazelcastServerInstance, JdbcTemplate jdbcTemplate) {
|
||||||
|
super(hazelcastServerInstance, jdbcTemplate);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String getCheckDbVersion() {
|
||||||
|
return CHECK_DB_VERSION;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Class[] getSerialVersionUIDClasses() {
|
||||||
|
return new Class[]{
|
||||||
|
ConstSerializable.class, ConstDictionarySerializable.class
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,41 @@
|
||||||
|
package ru.spcex.clearing.storage.services;
|
||||||
|
|
||||||
|
import com.hazelcast.core.EntryEvent;
|
||||||
|
import com.hazelcast.core.HazelcastInstance;
|
||||||
|
import com.hazelcast.query.Predicates;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
import ru.clearing.classes.StaticData.Company.Company;
|
||||||
|
import ru.clearing.classes.StaticData.Company.CompanyUpdate;
|
||||||
|
import ru.spcex.clearing.storage.utils.IMDGDistributedNames;
|
||||||
|
|
||||||
|
@Service
|
||||||
|
public class UpdateMapService extends AbstractUpdateMapService {
|
||||||
|
private final Logger log = LoggerFactory.getLogger(this.getClass());
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public UpdateMapService(HazelcastInstance hazelcastServerInstance, HazelcastLifecycleSupport lifecycleSupport) {
|
||||||
|
super(hazelcastServerInstance);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void addingListenersToCards() {
|
||||||
|
hazelcastServerInstance.getMap(IMDGDistributedNames.Map_Company).addLocalEntryListener(this, Predicates.alwaysTrue(), true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void entryModified(EntryEvent<Long, Object> event, String eventType) {
|
||||||
|
log.debug("{} {}", event, eventType);
|
||||||
|
Object value = (EVENT_DELETE.equals(eventType) ? event.getOldValue() : event.getValue());
|
||||||
|
if (value instanceof Company) {
|
||||||
|
CompanyUpdate companyUpdate = new CompanyUpdate();
|
||||||
|
createBusinessEvent(companyUpdate, eventType);
|
||||||
|
companyUpdate.setObject((Company) value);
|
||||||
|
hazelcastServerInstance.getMap(IMDGDistributedNames.Map_CompanyUpdate).put(companyUpdate.getId(), companyUpdate);
|
||||||
|
} else {
|
||||||
|
log.error("unexpected event {}", event);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Add table
Reference in a new issue