Блог

Я пытаюсь прочитать файл из GCP на основе уведомления, полученного в соответствии с потоком, определенным ниже:

Средство чтения файлов — Десериализует данные в сбор и отправляет для маршрутизации.

Я удаляю данные из коллекции объектов и отправляю их маршрутизатору для дальнейшей обработки. Поскольку у меня нет контроля над размером файла, я думаю о каком-то подходе к пакетной обработке процесса чтения.

В настоящее время активатор службы чтения файлов возвращает все Collection десериализованные объекты.

Вопрос:

• В случае, если я получу файл большего размера, т. Е. с записями 200 тыс., я хочу отправлять его пакетами на маршрутизатор значений заголовка, а не в коллекцию из 200 тыс. объектов.
• Если я конвертирую средство чтения файлов в разделитель и добавлю агрегатор после этого уведомления- > средство чтения файлов — > > агрегатор- > > > маршрутизатор. Мне все равно нужно было бы вернуть коллекцию всех объектов, а не итератор.
• Я не хочу загружать все записи в коллекцию.

Обновленный подход:

    public <S> Collection<S> readData(DataInfo dataInfo, Class<S> clazz) {
        Resource gcpResource = context.getResource("classpath://data.json")
        var tempDataSet = new HashSet<S>();
        AtomicInteger pivot = new AtomicInteger();
        try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(gcpResource.getInputStream()))) {
            bufferedReader.lines().map((dataStr) -> {
                try {
                    var data = deserializeData(dataStr, clazz);
                    return data;
                } catch (JsonProcessingException ex) {
                    throw new CustomException("PARSER-1001", "Error occurred while parsing", ex);
                }
            }).forEach(data -> {
                        if (BATCH_SIZE == pivot.get()) {
                            //When the size in tempDataSet reached BATCH_SIZE send the data in routing channel and reset the pivot
                            var message = MessageBuilder.withPayload(tempDataSet.clone())
                                    .setHeader(AppConstants.EVENT_HEADER_KEY, eventType)
                                    .build();
                            routingChannel.send(message);
                            pivot.set(0);
                            tempDataSet.removeAll(tempDataSet);
                        } else {
                            pivot.addAndGet(1);
                            tempDataSet.add(data);
                        }
                    });
            return tempDataSet;
        } catch (Exception ex) {
            throw new CustomException("PARSER-1002", "Error occurred while parsing", ex);
        }
    }
 

Если размер пакета в 100 и мы получили 1010 объектов. Будут созданы 11 пакетов, 10 со 100 и последний с 10 объектами в нем.

В случае, если я использую разделитель и передам ему поток, будет ли он ждать завершения всего потока, а затем отправлять собранную коллекцию, или мы сможем достичь чего-то близкого к предыдущему подходу, используя его?

Не уверен, в чем вопрос, но я бы выбрал FileSplitter Aggregator решение. Первый из них предназначен именно для использования при чтении потоковых файлов. Второй позволяет буферизировать входящие сообщения до тех пор, пока они не достигнут определенного состояния, поэтому он может выдавать одно сообщение вниз по потоку. Это сообщение действительно может быть с коллекцией в качестве полезной нагрузки.

Вот их документы для вашего рассмотрения:

https://docs.spring.io/spring-integration/docs/current/reference/html/message-routing.html#aggregator

https://docs.spring.io/spring-integration/docs/current/reference/html/file.html#file-splitter