передача потока Akka в вышестоящий сервис для заполнения

9

Мне нужно вызвать вышестоящую службу (службу BLOB-объектов Azure), чтобы отправить данные в OutputStream, который затем мне нужно развернуть и отправить обратно клиенту через akka. Без akka (и просто кода сервлета) я просто получил бы ServletOutputStream и передал его методу службы Azure.

Самое близкое, на что я могу попытаться наткнуться, и, очевидно, это неправильно, это что-то вроде этого

        Source<ByteString, OutputStream> source = StreamConverters.asOutputStream().mapMaterializedValue(os -> {
            blobClient.download(os);
            return os;
        });

        ResponseEntity resposeEntity = HttpEntities.create(ContentTypes.APPLICATION_OCTET_STREAM, preAuthData.getFileSize(), source);

        sender().tell(new RequestResult(resposeEntity, StatusCodes.OK), self());

Идея в том, что я вызываю вышестоящий сервис для получения выходного потока, вызывая blobClient.download (os);

Кажется, что лямбда-функция вызывается и возвращается, но затем она перестает работать, потому что нет данных или чего-то еще. Как будто я не должен иметь эту лямбда-функцию, но возможно вернуть какой-то объект, который работает? Точно сказать не могу.

Как это сделать?

MeBigFatGuy
источник
Каково поведение download? Потоковые данные передаются osи возвращаются только после записи данных?
Alec

Ответы:

2

Настоящая проблема здесь заключается в том, что API Azure не предназначен для противодействия давлению. Выходной поток не может сообщить в Azure о том, что он не готов для получения дополнительных данных. Другими словами: если Azure отправляет данные быстрее, чем вы можете их использовать, где-то должен произойти какой-то ужасный сбой переполнения буфера.

Принимая этот факт, мы можем сделать следующее:

  • Используйте, Source.lazySourceчтобы начать загрузку данных только тогда, когда есть потребность в нисходящем потоке (он же источник запускается и данные запрашиваются).
  • Поместите downloadвызов в другой поток, чтобы он продолжал выполняться, не блокируя возвращение источника. Один из способов сделать это с помощью Future(я не уверен, каковы лучшие практики Java, но в любом случае должен работать нормально). Хотя изначально это не имеет значения, вам может потребоваться выбрать контекст выполнения, отличный от system.dispatcher- все зависит от того download, блокирует он или нет.

Я заранее прошу прощения, если этот код Java искажен - я использую Akka со Scala, так что это все из рассмотрения справочника по синтаксису Java API и Akka.

ResponseEntity responseEntity = HttpEntities.create(
  ContentTypes.APPLICATION_OCTET_STREAM,
  preAuthData.getFileSize(),

  // Wait until there is downstream demand to intialize the source...
  Source.lazySource(() -> {
    // Pre-materialize the outputstream before the source starts running
    Pair<OutputStream, Source<ByteString, NotUsed>> pair =
      StreamConverters.asOutputStream().preMaterialize(system);

    // Start writing into the download stream in a separate thread
    Futures.future(() -> { blobClient.download(pair.first()); return pair.first(); }, system.getDispatcher());

    // Return the source - it should start running since `lazySource` indicated demand
    return pair.second();
  })
);

sender().tell(new RequestResult(responseEntity, StatusCodes.OK), self());
сельдь
источник
Фантастический. Спасибо большое. Небольшое изменение вашего примера: Futures.future (() -> {blobClient.download (pair.first ()); return pair.first ();}, system.getDispatcher ());
MeBigFatGuy
@MeBigFatGuy Правильно, спасибо!
Алек
1

В OutputStreamэтом случае это «материализованное значение» Sourceи оно будет создано только после запуска потока (или «материализации» в текущий поток). Запуск его вне вашего контроля, так как вы передаете в SourceAkka HTTP, и это позже фактически запустит ваш источник.

.mapMaterializedValue(matval -> ...)обычно используется для преобразования материализованного значения, но поскольку оно вызывается как часть материализации, вы можете использовать его для выполнения побочных эффектов, таких как отправка матвала в сообщении, как вы уже поняли, в этом нет ничего плохого что даже если это выглядит прикольно. Важно понимать, что поток не завершит свою материализацию и начнет работать, пока эта лямбда не завершится. Это означает проблемы, если download()блокировать, а не завершать какую-то работу в другом потоке и немедленно возвращать.

Однако существует другое решение: Source.preMaterialize()оно материализует источник и дает вам Pairматериализованное значение и новое, Sourceкоторое можно использовать для потребления уже запущенного источника:

Pair<OutputStream, Source<ByteString, NotUsed>> pair = 
  StreamConverters.asOutputStream().preMaterialize(system);
OutputStream os = pair.first();
Source<ByteString, NotUsed> source = pair.second();

Обратите внимание, что в вашем коде есть еще несколько вещей, о которых следует подумать, особенно если blobClient.download(os)вызов блокируется до тех пор, пока он не будет завершен, и вы вызываете его от актера, в этом случае вы должны убедиться, что ваш актер не остановит диспетчер и не остановит его. другие действующие лица в вашем приложении от выполнения (см. документы Akka: https://doc.akka.io/docs/akka/current/typed/dispatchers.html#blocking-needs-careful-management ).

johanandren
источник
1
Спасибо за ответ. Я не понимаю, как это может сработать? Куда идут байты при вызове blobClient.download (os) (если я вызываю его сам)? Представьте, что есть терабайт данных, ожидающих записи. мне кажется, что вызов blobClient.download должен вызываться из вызова sender.tell, так что это в основном операция типа IOUtils.copy. Используя preMaterialize, я не вижу, как это происходит?
MeBigFatGuy
OutputStream имеет внутренний буфер, он начнет принимать записи до тех пор, пока этот буфер не заполнится, если асинхронный нисходящий поток не начнет потреблять элементы, тогда он будет блокировать поток записи (именно поэтому я упомянул, что важно обрабатывать блокировку).
Йоханандрен
1
Но если я preMaterialize и получаю OutputStream, то это мой код, который выполняет blobClient.download (os); правильный? Это означает, что он должен завершиться, прежде чем я смогу продолжить, что невозможно.
MeBigFatGuy
Если загрузка (os) не является ветвью потока, вам придется иметь дело с блокировкой и убедиться, что она не останавливает какую-либо другую операцию. Один из способов - создать поток для выполнения работы, другой - сначала ответить от актера, а затем выполнить блокировку, в этом случае вы должны убедиться, что актер не истощает других акторов, см. Ссылку в конце мой ответ.
Йоханандрен
на данный момент я просто пытаюсь заставить его работать на всех. Он даже не может обработать 10-байтовый файл.
MeBigFatGuy