From 1dee51257ec572c1a9a6e222a048f65c3f9fabcb Mon Sep 17 00:00:00 2001 From: akurakin Date: Sun, 3 Jul 2022 23:07:50 +0300 Subject: [PATCH] =?UTF-8?q?http://jira.mfd.msk:8088/browse/CLS-1=20storage?= =?UTF-8?q?=20=D0=BA=D0=BE=D0=BF=D0=B8=D0=BF=D0=B0=D1=81=D1=82=20=D1=81?= =?UTF-8?q?=D0=B5=D1=80=D0=B2=D0=B8=D1=81=D0=BE=D0=B2=20(=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D0=BE=D1=80=D0=BE=D0=B6=D0=BD=D0=BE-=D1=82=D1=83=D0=B4?= =?UTF-8?q?=D1=83=D1=88=D0=BA=D0=B8=20=D0=B8=20=D0=BA=D0=BE=D0=BD=D1=84?= =?UTF-8?q?=D0=B8=D0=B3=D0=B0=20=D0=BD=D0=B5=D1=82):=20=D0=B3=D0=BE=D1=82?= =?UTF-8?q?=D0=BE=D0=B2=D1=8B=20=D0=BD=D0=B0=2098%?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AbstractHazelcastLifecycleSupport.java | 216 ++++++++++++++++++ .../services/AbstractUpdateMapService.java | 62 +++++ .../services/HazelcastLifecycleSupport.java | 37 +++ .../storage/services/UpdateMapService.java | 41 ++++ 4 files changed, 356 insertions(+) create mode 100644 storage/src/main/java/ru/spcex/clearing/storage/services/AbstractHazelcastLifecycleSupport.java create mode 100644 storage/src/main/java/ru/spcex/clearing/storage/services/AbstractUpdateMapService.java create mode 100644 storage/src/main/java/ru/spcex/clearing/storage/services/HazelcastLifecycleSupport.java create mode 100644 storage/src/main/java/ru/spcex/clearing/storage/services/UpdateMapService.java diff --git a/storage/src/main/java/ru/spcex/clearing/storage/services/AbstractHazelcastLifecycleSupport.java b/storage/src/main/java/ru/spcex/clearing/storage/services/AbstractHazelcastLifecycleSupport.java new file mode 100644 index 000000000..15f90b64c --- /dev/null +++ b/storage/src/main/java/ru/spcex/clearing/storage/services/AbstractHazelcastLifecycleSupport.java @@ -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> 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 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> results = executor.invokeAll(tasks); + long maxKey = 0L; + for (Future 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; + } +} diff --git a/storage/src/main/java/ru/spcex/clearing/storage/services/AbstractUpdateMapService.java b/storage/src/main/java/ru/spcex/clearing/storage/services/AbstractUpdateMapService.java new file mode 100644 index 000000000..744416d51 --- /dev/null +++ b/storage/src/main/java/ru/spcex/clearing/storage/services/AbstractUpdateMapService.java @@ -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, EntryUpdatedListener, EntryRemovedListener { + 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 event) { + entryModified(event, EVENT_CREATE); + } + + @Override + public void entryUpdated(EntryEvent event) { + entryModified(event, EVENT_UPDATE); + } + + @Override + public void entryRemoved(EntryEvent event) { + entryModified(event, EVENT_DELETE); + } + + protected abstract void entryModified(EntryEvent 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); + } +} diff --git a/storage/src/main/java/ru/spcex/clearing/storage/services/HazelcastLifecycleSupport.java b/storage/src/main/java/ru/spcex/clearing/storage/services/HazelcastLifecycleSupport.java new file mode 100644 index 000000000..d755f85d2 --- /dev/null +++ b/storage/src/main/java/ru/spcex/clearing/storage/services/HazelcastLifecycleSupport.java @@ -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 + }; + } +} \ No newline at end of file diff --git a/storage/src/main/java/ru/spcex/clearing/storage/services/UpdateMapService.java b/storage/src/main/java/ru/spcex/clearing/storage/services/UpdateMapService.java new file mode 100644 index 000000000..63e50b394 --- /dev/null +++ b/storage/src/main/java/ru/spcex/clearing/storage/services/UpdateMapService.java @@ -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 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); + } + } +}