From 7986fc002dae5c520a4b84cd805f6b7e9f02f4c0 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Thu, 17 Jul 2025 11:17:56 +0300 Subject: [PATCH] Decrease memory usage for loading and creating reports --- .../credit_tracker/facade/SignalFacade.java | 4 +- .../service/MonitoringService.java | 5 +- .../credit_tracker/service/SignalService.java | 40 +++++++++++--- .../service/UpdatingFlagsService.java | 4 +- .../service/signal/SignalRestService.java | 52 +++++++++++------- .../xml/stax/SignalsStaxXmlWriter.java | 53 ++++++++++++++++++- 6 files changed, 125 insertions(+), 33 deletions(-) diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index 75f4ebd..eee713c 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -18,7 +18,6 @@ import ru.nbch.credit_tracker.entities.response.active_packages.Packages; import ru.nbch.credit_tracker.entities.response.success.MonitoringPackageDeleteResponse; import ru.nbch.credit_tracker.entities.response.success.MonitoringReportListResponse; import ru.nbch.credit_tracker.entities.response.success.MonitoringResponse; -import ru.nbch.credit_tracker.entities.signal_package.Signals; import ru.nbch.credit_tracker.enums.SignalError; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.enums.XmlErrorFormat; @@ -163,8 +162,7 @@ public class SignalFacade { signalService.registerSubjects(subjects, packageId); subjects.clear(); signalService.updateSubjects(packageId, null); - Signals signals = signalService.buildSignals(packageId, null); - if (signalRestService.update(signals)) { + if (signalRestService.update(packageId, null)) { log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName); signalService.updatePackageStatus(packageId, Status.ON_MONITORING); } else { diff --git a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java index 6e6ae20..7d25c2c 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java @@ -1,6 +1,7 @@ package ru.nbch.credit_tracker.service; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; @@ -82,14 +83,14 @@ public class MonitoringService { LocalDateTime lastDownloadedReportDate = CTUtil.parseDateFromReportFileName(userPackage.getLastDownloadedReport()); - if (userPackage.getLastDownloadedReport() != null) { + if (StringUtils.isNotBlank(userPackage.getLastDownloadedReport())) { log.info("Last downloaded report found for package {}. Last Downloaded Report Name: {}", userPackage.getPackageName(), userPackage.getLastDownloadedReport()); } else { log.info("Last Downloaded Report Name is empty"); } OnlineDownloadReport report = null; - if (userPackage.getLastDownloadedReport() == null && newReport.isPresent()) { + if (StringUtils.isBlank(userPackage.getLastDownloadedReport()) && newReport.isPresent()) { report = loadReport(newReport.get()); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным } else if (lastDownloadedReportDate != null && newReport.isPresent()) { LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport.get()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java index 48fc786..6a7fd91 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/SignalService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/SignalService.java @@ -19,9 +19,15 @@ import ru.nbch.credit_tracker.service.indic.PhoneService; import ru.nbch.credit_tracker.service.signal.HitsService; import ru.nbch.credit_tracker.service.signal.PhoneFidMapService; import ru.nbch.credit_tracker.service.signal.UserPackageService; +import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; import java.time.LocalDateTime; import java.util.*; +import java.util.function.Consumer; import java.util.function.Predicate; @Slf4j @@ -165,7 +171,32 @@ public class SignalService { userPackageService.updatePackageStatus(packageId, status); } - public Signals buildSignals(Integer packageId, Predicate filter) { + public String writeSignals(Integer packageId, Predicate filter) { + String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator + packageId + File.separator; + String filePath = folderPath + UUID.randomUUID() + "_output.xml"; + + File folder = new File(folderPath); + if (!folder.exists()) { + try { + Files.createDirectories(Path.of(folderPath)); + } catch (IOException e) { + log.error("Could not create folder: {}", folderPath, e); + throw new RuntimeException(e); + } + } + + Signals signals = buildSignals(packageId); + + SignalsStaxXmlWriter signalsStaxXmlWriter = new SignalsStaxXmlWriter("windows-1251"); + + signalsStaxXmlWriter.writeWithStax(filePath, + signals, + writer -> buildPersons(packageId, filter, writer)); + + return filePath; + } + + private Signals buildSignals(Integer packageId) { UserPackage userPackage = userPackageService.findPackage(packageId); Signals signals = new Signals(); signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы) @@ -174,13 +205,11 @@ public class SignalService { auth.setUser(config.getSignalsuser()); signals.setAuth(auth); - signals.getConditions().getPersons().getPerson().addAll(buildPersons(packageId, null)); return signals; } - private List buildPersons(Integer packageId, Predicate filter) { - List result = new ArrayList<>(); + private void buildPersons(Integer packageId, Predicate filter, Consumer> writer) { int offset = 0; List batch; do { @@ -192,12 +221,11 @@ public class SignalService { persons = findPersons(batch); } if (!persons.isEmpty()) { - result.addAll(persons); + writer.accept(persons); } offset += batchSize; } while (!batch.isEmpty()); - return result; } private List findPersons(List subjects) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index bf65f76..5bfc305 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -6,7 +6,6 @@ import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import ru.nbch.credit_tracker.entities.dto.UserPackage; -import ru.nbch.credit_tracker.entities.signal_package.Signals; import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.service.signal.SignalRestService; @@ -57,8 +56,7 @@ public class UpdatingFlagsService { Predicate filter = subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt()); boolean updated = signalService.updateSubjects(packageId, filter); if (updated) { - Signals signals = signalService.buildSignals(packageId, filter); - if (signalRestService.update(signals)) { + if (signalRestService.update(packageId, filter)) { log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName); signalService.updatePackageStatus(packageId, Status.ON_MONITORING); } else { diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java index 845c287..09ab441 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/SignalRestService.java @@ -3,7 +3,7 @@ package ru.nbch.credit_tracker.service.signal; import jakarta.xml.bind.DatatypeConverter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.FileUrlResource; import org.springframework.core.io.Resource; import org.springframework.http.*; import org.springframework.stereotype.Service; @@ -11,17 +11,22 @@ import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.client.RestTemplate; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.dto.PhoneFidMap; import ru.nbch.credit_tracker.entities.response.external_errors.SgnlReport; import ru.nbch.credit_tracker.entities.response.online_report.SgnlOnlineReport; -import ru.nbch.credit_tracker.entities.signal_package.Signals; +import ru.nbch.credit_tracker.service.SignalService; import ru.nbch.credit_tracker.service.xml.jaxb.IXmlProcessor; -import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter; import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor; -import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.security.MessageDigest; import java.util.Base64; +import java.util.function.Predicate; import java.util.zip.GZIPOutputStream; @Slf4j @@ -36,20 +41,24 @@ public class SignalRestService { private final RestTemplate restTemplate; private final IXmlProcessor xmlProcessor; private final StaxXmlProcessor staxProcessor; + private final SignalService signalService; private final String authToken; public SignalRestService(@Qualifier(value = "restTemplate") RestTemplate restTemplate, IXmlProcessor xmlProcessor, StaxXmlProcessor staxProcessor, + SignalService signalService, CreditTrackerSettings config) throws Exception { this.restTemplate = restTemplate; this.xmlProcessor = xmlProcessor; this.staxProcessor = staxProcessor; + this.signalService = signalService; this.authToken = buildToken(config.getSignalsToken(), config.getSignalsuser()); } - public boolean update(Signals signals) throws Exception { - Resource request = compressFile(signals); + public boolean update(Integer packageId, Predicate filter) throws Exception { + String path = signalService.writeSignals(packageId, filter); + Resource request = compressFile(path); MultiValueMap body = new LinkedMultiValueMap<>(); body.add("file", request); @@ -64,6 +73,12 @@ public class SignalRestService { if (response.getStatusCode().value() == 302) { // good response + try { + Files.delete(Path.of(path)); + Files.delete(Path.of(path + ".gz")); + } catch (IOException e) { + log.warn("File {} hasn't been deleted", path); + } return true; } else if (response.getStatusCode().value() == 200 && response.getBody() != null) { SgnlReport error = staxProcessor.read(response.getBody(), SgnlReport.class); @@ -72,6 +87,7 @@ public class SignalRestService { } else { log.error("неизвестная ошибка при постановке пакета на мониторинг в официальный сервис сигналов: {}", response.getBody()); } + log.info("Files will not be deleted: {}, {}", path, path + ".gz"); return false; } return false; @@ -171,20 +187,20 @@ public class SignalRestService { } } - private Resource compressFile(Signals signals) throws Exception { - SignalsStaxXmlWriter signalsStaxXmlWriter = new SignalsStaxXmlWriter("windows-1251"); + public Resource compressFile(String path) throws Exception { + try (InputStream in = Files.newInputStream(Path.of(path)); + OutputStream out = Files.newOutputStream(Path.of(path + ".gz")); + GZIPOutputStream gzipOut = new GZIPOutputStream(out)) { - ByteArrayOutputStream byteStream = new ByteArrayOutputStream(); - try (GZIPOutputStream gzipStream = new GZIPOutputStream(byteStream)) { - signalsStaxXmlWriter.writeWithStax(gzipStream, signals, false); - } - - return new ByteArrayResource(byteStream.toByteArray()) { - @Override - public String getFilename() { - return "arhive.gz"; + byte[] buffer = new byte[65536]; + int bytesRead; + while ((bytesRead = in.read(buffer)) > 0) { + gzipOut.write(buffer, 0, bytesRead); } - }; + gzipOut.finish(); + + return new FileUrlResource(path + ".gz"); + } } private boolean isXmlError(byte[] data) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java index 3ddf844..aded3ad 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java +++ b/src/main/java/ru/nbch/credit_tracker/service/xml/stax/SignalsStaxXmlWriter.java @@ -7,15 +7,63 @@ import javax.xml.stream.XMLEventFactory; import javax.xml.stream.XMLEventWriter; import javax.xml.stream.XMLOutputFactory; import java.io.BufferedOutputStream; +import java.io.FileOutputStream; import java.io.OutputStream; +import java.util.List; +import java.util.function.Consumer; @Slf4j -public class SignalsStaxXmlWriter extends StaxXmlWriter{ +public class SignalsStaxXmlWriter extends StaxXmlWriter { public SignalsStaxXmlWriter(String charsetName) { super(charsetName); } + public void writeWithStax(String filePath, Signals signals, LoaderWriter loader) { + try (FileOutputStream out = new FileOutputStream(filePath)) { + XMLOutputFactory output = XMLOutputFactory.newInstance(); + XMLEventFactory xmlEventFactory = XMLEventFactory.newInstance(); + XMLEventWriter writer = output.createXMLEventWriter(out, charsetName); + writer = new IndentingXMLEventWriter(writer); + + writerWrapper = new WriterWrapper(xmlEventFactory, writer); + writerWrapper.addStartDocument(charsetName); + + writerWrapper.addStartElement("Signals"); + writerWrapper.addAttribute("online", signals.getOnline()); + writerWrapper.addSimplelement("Id", signals.getId()); + writeAuth(signals.getAuth()); + + writerWrapper.addStartElement("Conditions"); + writerWrapper.addStartElement("Persons"); + + + loader.accept(persons -> { + try { + writePersons(persons); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + + writerWrapper.addEndElement("Persons"); + writerWrapper.addEndElement("Conditions"); + + writerWrapper.addEndElement("Signals"); + + writerWrapper.addEndDocument(); + writerWrapper.flush(); + System.gc(); + } catch (Throwable e) { + log.error("Error serializing object to XML string: {}", e.getMessage(), e); + throw new IllegalStateException("cannot serialize object to XML"); + } + } + + public void writePersons(List persons) throws Exception{ + writerWrapper.writeCollection("p", persons, this::writePerson); + } + public void writeWithStax(OutputStream outputStream, Signals signals, boolean useStandalone) { try (BufferedOutputStream bos = new BufferedOutputStream(outputStream, 4 * 1024 * 1024)) { @@ -110,4 +158,7 @@ public class SignalsStaxXmlWriter extends StaxXmlWriter{ } + public interface LoaderWriter { + void accept(Consumer> writer); + } }