[JAVA] Ich möchte die Aggregationsverarbeitung mit Spring-Batch durchführen

Hintergrund

In einer mit Spring-Boot erstellten WEB-Anwendung Ich möchte einen solchen Stapelprozess implementieren, der den Inhalt der vorbereiteten API-Ausführungsprotokolldatei aggregiert und das Ergebnis in der Datenbank speichert. Es gibt jedoch keine Website, die als Referenz für die Umsetzung dieser Anforderung verwendet werden könnte, und sie war lächerlich verstopft, sodass ich sie als Memorandum belassen werde. (Ich weiß nicht, ob das das Beste ist)

• Ungefähr 7 Tage Chargenhistorie zum Zeitpunkt der Veröffentlichung durch uns

Hauptanforderungen

• Ich möchte für jedes bestimmte Element in den Daten aggregieren
• Die Aggregation erfolgt auf der Java-Seite (nicht in SQL)
• Konvertieren Sie einen Teil der Daten in eine Zeichenfolge, die vor dem Speichern in der Datenbank zugeordnet wurde

Vorausgesetzte Daten (Beispiel)

Vorher: Protokollformat, das aggregiert werden soll

Die URL / HTTP-Methode / HTTP-Statuscode / Ausführungszeit / Ausführungsdatum und -zeit zum Zeitpunkt des Zugriffs ist im TSV-Format aufgeführt. Beispiel: / api / ・ ・ ・ GET 200 48 2020/08/14 11:05:42 701 / api / ET ・ ・ GET 200 27 2020/08/14 11:05:43 352 / api / ・ ・ ・ / 41 DELETE 401 10 2020/08/14 11:05:46 780 / api / ・ ・ ・ / 42 PUT 200 108 2020/08/14 11:06:16 824 / api / ST ・ ・ POST 500 806 2020/08/14 11:06:30 252 ・ ・ ・

Nachher: Format beim Speichern der Datenbank

• Verarbeite URL in einen logischen API-Namen
• Aggregieren Sie "Zugriffsanzahl" und "durchschnittliche Ausführungszeit" unter Verwendung von "URL (API-Name) / HTTP-Methode / HTTP-Statuscode" als Schlüssel
• Datensatzformat: API-Name / HTTP-Methode / HTTP-Statuscode / Anzahl der Zugriffe / durchschnittliche Ausführungszeit / Gesamtdatum und -zeit
• Beispiel: ○○ API GET 200 10 240

Implementierung

Politik

• Implementiert mit Chunk Model
• Reader: Zeile für Zeile lesen (1 Datenzeile = 1Element)
• Prozessor: Verarbeitung von vom Reader gesendeten Elementen
• Writer: Aggregieren Sie die Liste der verarbeiteten Elemente, die vom Prozessor gesendet wurden → Bereiten Sie eine andere Writer-Instanz vor und übergeben Sie das aggregierte Ergebnis an diese

Punkt

• Führen Sie die Aggregationsverarbeitung mit der Stream-API durch
• Bereiten Sie zwei vor, einen Writer, der die Elementgruppe vom Prozessor empfängt, und einen Writer, der den Schreibprozess tatsächlich ausführt.

Code

• Die Verarbeitung um Batch-Einstellungen und Inhalte der Entitätsklasse entfällt.

Dto Klasse

@Data
public class LogCollectedDto {
  //API-Name
  private String apiName;
  //HTTP-Methode
  private String httpMethod;
  //HTTP-Statuscode
  private String httpCode;
  //Ausführungszeit(ms)
  private String executionTime;
  //Datum und Uhrzeit der Aggregation
  private String collectedDate;
}

Reader

In Bean definiert

  @Bean
  public FlatFileItemReader<LogCollectedDto> reader() {

    final String READ_FILE_PATH = <Protokolldateiname zum Lesen>;

    FlatFileItemReader<LogCollectedDto> reader = new FlatFileItemReader<>();
    reader.setResource(new FileSystemResource(READ_FILE_PATH));
    reader.setEncoding(StandardCharsets.UTF_8.name());
    reader.setLinesToSkip(0);
    reader.setLineMapper(
        new DefaultLineMapper() {
          {
            setLineTokenizer(
                new DelimitedLineTokenizer(DelimitedLineTokenizer.DELIMITER_TAB) {
                  {
                    setNames(
                        new String[] {
                          "apiUrl", "httpMethod", "httpCode", "executionTime", "collectedDate"
                        });
                  }
                });
            setFieldSetMapper(
                new BeanWrapperFieldSetMapper<LogCollectedDto>() {
                  {
                    setTargetType(LogCollectedDto.class);
                  }
                });
          }
        });
    return reader;
  }

Processor

In eine andere Klasse ausschneiden

public class CustomItemProcessor implements ItemProcessor<LogCollectedDto, LogCollectedDto> {

  @Override
  public LogCollectedDto process(LogCollectedDto item) throws Exception {

    //Validierung des abgerufenen Elements, überspringen Sie diese Zeile, wenn sie falsch ist
    if (!validateItem(item)) {
      return null;
    }

    //Halten Sie den erworbenen Gegenstand einmal in einer anderen Variablen, um ihn später zu verarbeiten
    //(Wenn Sie das Argument direkt ändern, wird das beim Unterbrechen und Neustarten erfasste Element möglicherweise zu den verarbeiteten Daten.)
    LogCollectedDto afterItem = item;
    //Dateninhaltsverarbeitung (separate Methode weggelassen)
    afterItem.setApiName(getApiName(・ ・ ・));

    return afterItem;
  }

//(Weggelassen)

}

Writer

In eine andere Klasse ausschneiden

@RequiredArgsConstructor
public class CustomItemWriter extends JpaItemWriter<LogCollectedDto> {

  private final JpaItemWriter<Log> jpaItemWriter;

  @Override
  public void write(List<? extends LogCollectedDto> items) {

   //Aggregieren Sie die vom Prozessor empfangenen Elemente und übergeben Sie sie an eine andere Writer-Instanz
    Map<String, List<LogCollectedDto>> groupingMap = groupingItems(items);
    jpaItemWriter.write(collectItems(groupingMap));
  }

  /**
   *Gruppieren von vom Prozessor empfangenen Elementen
   *API-Name und HTTP-Status als zusammengesetzter Schlüssel
   *
   * @param list
   * @return
   */
  private Map<String, List<LogCollectedDto>> groupingItems(List<? extends LogCollectedDto> list) {
    //Erstellen Sie einen zusammengesetzten Schlüssel
    Function<LogCollectedDto, String> compositeKey =
        logDto -> {
          StringBuffer sb = new StringBuffer();
          sb.append(logDto.getApiName()).append("-").append(logDto.getHttpMethod());
          return sb.toString();
        };

    Map<String, List<LogCollectedDto>> grpByComplexKeys =
        list.stream().collect(Collectors.groupingBy(compositeKey));

    return grpByComplexKeys;
  }

  /**
   *Generieren Sie eine Liste von Entitäten, indem Sie gruppierte Elemente zusammenfassen
   *
   * @param groupingMap
   * @return
   */
  private List<Log> collectItems(Map<String, List<LogCollectedDto>> groupingMap) {

    List<Log> recordList = new ArrayList<>();

    for (List<LogCollectedDto> dtoList : groupingMap.values()) {
      //Instanziierung von Entitätsklassen
      Log record = new Log();
      //Aggregationsverarbeitung
      record.setApiName(dtoList.stream().findFirst().get().getApiName());
      record.setHttpCode(dtoList.stream().findFirst().get().getHttpCode());
      record.setHttpMethod(dtoList.stream().findFirst().get().getHttpMethod());
      record.setAccesses(dtoList.size());
      record.setAverageTime(
          dtoList.stream()
              .collect(
                  Collectors.averagingDouble(dto -> Double.parseDouble(dto.getExecutionTime()))));
      record.setCollectedTime(LocalDateTime.now());
      recordList.add(record);
    }

    return recordList;
  }
}

Besorgnis, Sorge

• Mit nur der obigen Implementierung entspricht die Anzahl der Zeilen, die gleichzeitig aggregiert werden können, der festgelegten Blockgröße.
• → Nehmen Sie das Obige als einen Schritt, und wenn im nächsten Schritt derselbe Schlüssel in den in der Datenbank gespeicherten Inhalten vorhanden ist, sollte er zum Nachzählen implementiert werden (die DB-Last scheint hoch zu sein ...)
• Ich kann das Gefühl nicht leugnen, dass die Verantwortlichkeiten des Schriftstellers überladen sind.

Zusammenfassung

Es gibt Bedenken, aber ich persönlich denke, es ist die sauberste Art zu schreiben. Wir sind immer auf der Suche nach Meinungen. Fazit: Stream API ist die stärkste! !! !!