Eu tenho um Akka Stream e quero que ele envie mensagens no fluxo aproximadamente a cada segundo.

Tentei duas maneiras de resolver esse problema: a primeira maneira era fazer o produtor no início do fluxo enviar mensagens apenas uma vez a cada segundo quando uma mensagem Continue entra nesse ator.

// When receive a Continue message in a ActorPublisher // do work then... if (totalDemand > 0) { import scala.concurrent.duration._ context.system.scheduler.scheduleOnce(1 second, self, Continue) }

Isso funciona por um curto período de tempo e, em seguida, uma enxurrada de mensagens Continue aparece no ator ActorPublisher, suponho (acho, mas não tenho certeza) do downstream por meio de contrapressão solicitando mensagens, pois o downstream pode consumir rapidamente, mas o upstream não está produzindo a uma taxa rápida . Portanto, este método falhou.

A outra maneira que tentei foi através do controle de contrapressão, usei umMaxInFlightRequestStrategy noActorSubscriber no final do fluxo para limitar o número de mensagens a 1 por segundo. Isso funciona, mas as mensagens são recebidas aproximadamente às três, mas não apenas uma de cada vez. Parece que o controle de contrapressão não altera imediatamente a taxa de mensagens que chegam OU as mensagens já estavam na fila no fluxo e aguardando processamento.

Portanto, o problema é: como posso ter um Akka Stream que pode processar uma mensagem apenas por segundo?

Eu descobri issoMaxInFlightRequestStrategy é uma maneira válida de fazer isso, mas devo definir o tamanho do lote como 1; o tamanho do lote é o padrão 5, o que estava causando o problema encontrado. Também é uma maneira muito complicada de resolver o problema agora que estou vendo a resposta enviada aqui.