Problemas no processamento em lote de solicitações e suas soluções (parte 1)

Quase todos os produtos de software modernos consistem em vários serviços. Frequentemente, os longos tempos de resposta do canal entre serviços se tornam uma fonte de problemas de desempenho. A solução padrão para esse tipo de problema é compactar várias solicitações interserviços em um pacote, chamado de lote.

Se você usar o processamento em lote, poderá não estar satisfeito com o resultado em termos de desempenho ou compreensibilidade do código. Esse método não é tão fácil para o chamador quanto você imagina. Para diferentes propósitos e em diferentes situações, as decisões podem variar bastante. Em exemplos específicos, mostrarei os prós e contras de várias abordagens.

Projeto de demonstração

Para maior clareza, considere um exemplo de um dos serviços no aplicativo em que estou trabalhando atualmente.


Existe um serviço de aprovação de documentos. Alguém cria um documento e o envia para discussão, durante o qual são feitas edições e, finalmente, o documento é consistente. O serviço de reconciliação em si não sabe nada sobre documentos: é apenas uma conversa de coordenadores com pequenas funções adicionais, que não consideraremos aqui.

Portanto, há salas de bate-papo (correspondentes a documentos) com um conjunto predefinido de participantes em cada um deles. Como nos bate-papos regulares, as mensagens contêm texto e arquivos e podem ser respostas e encaminhamentos:

data class ChatMessage (
   // nullable persist
   val id : Long ? = null ,
   /** */
   val author : UserReference ,
   /** */
   val message : String ,
   /** */
   // - JPA+ null,
   val files : List < FileReference > ? = null ,
   /** , */
   val replyTo : ChatMessage ? = null ,
   /** , */
   val forwardFrom : ChatMessage ? = null
)

Links para o arquivo e usuário são links para outros domínios. Ele mora conosco assim:

typealias FileReference = Long
typealias UserReference = Long

Os dados do usuário são armazenados no Keycloak e recuperados via REST. O mesmo vale para arquivos: arquivos e meta-informações sobre eles residem em um serviço de armazenamento de arquivos separado.

Todas as chamadas para esses serviços são solicitações pesadas . Isso significa que a sobrecarga para transportar essas solicitações é muito maior do que o tempo necessário para serem processadas por um serviço de terceiros. Nas bancadas de teste, o tempo de chamada típico para esses serviços é de 100 ms; portanto, no futuro, usaremos esses números.

Precisamos criar um controlador REST simples para receber as últimas N mensagens com todas as informações necessárias. Ou seja, acreditamos que no frontend, o modelo de mensagem é quase o mesmo e precisamos enviar todos os dados. A diferença entre o modelo de front-end é que o arquivo e o usuário precisam ser apresentados em um formato levemente descriptografado para torná-los links:

/** */
data class ReferenceUI (
   /** url */
   val ref : String ,
   /** */
   val name : String
)
data class ChatMessageUI (
   val id : Long ,
   /** */
   val author : ReferenceUI ,
   /** */
   val message : String ,
   /** */
   val files : List < ReferenceUI >,
   /** , */
   val replyTo : ChatMessageUI ? = null ,
   /** , */
   val forwardFrom : ChatMessageUI ? = null
)

Precisamos implementar o seguinte:

interface ChatRestApi {
   fun getLast ( n : Int ) : List < ChatMessageUI >
}

UI Postfix significa modelos de DTO para o frontend, ou seja, o que devemos fornecer através do REST.

Pode parecer surpreendente aqui que não passamos nenhum identificador de bate-papo e mesmo no modelo ChatMessage / ChatMessageUI não é. Fiz isso de propósito, para não confundir o código dos exemplos (os chats são isolados, para que possamos assumir que temos um).


Uma limitação importante: assumimos que não há longas cadeias de respostas ou encaminhamentos. Ou seja, eles são, mas seu comprimento não excede três mensagens. A cadeia de mensagens front-end deve ser transmitida na íntegra.

Para receber dados de serviços externos, existem APIs:

interface ChatMessageRepository {
   fun findLast ( n : Int ) : List < ChatMessage >
}
data class FileHeadRemote (
   val id : FileReference ,
   val name : String
)
interface FileRemoteApi {
   fun getHeadById ( id : FileReference ) : FileHeadRemote
   fun getHeadsByIds ( id : Set < FileReference > ) : Set < FileHeadRemote >
   fun getHeadsByIds ( id : List < FileReference > ) : List < FileHeadRemote >
   fun getHeadsByChat () : List < FileHeadRemote >
}
data class UserRemote (
   val id : UserReference ,
   val name : String
)
interface UserRemoteApi {
   fun getUserById ( id : UserReference ) : UserRemote
   fun getUsersByIds ( id : Set < UserReference > ) : Set < UserRemote >
   fun getUsersByIds ( id : List < UserReference > ) : List < UserRemote >
}

Pode-se observar que o processamento em lote foi inicialmente fornecido em serviços externos e em ambos os casos: através de Set (sem preservar a ordem dos elementos, com chaves exclusivas) e através de List (pode haver duplicatas - o pedido é preservado).

Implementações simples

Implementação ingênua

A primeira implementação ingênua do nosso controlador REST será semelhante a isso na maioria dos casos:

class ChatRestController (
   private val messageRepository : ChatMessageRepository ,
   private val userRepository : UserRemoteApi ,
   private val fileRepository : FileRemoteApi
) : ChatRestApi {
   override fun getLast ( n : Int ) =
     messageRepository . findLast ( n )
       . map { it . toFrontModel () }

   private fun ChatMessage . toFrontModel () : ChatMessageUI =
     ChatMessageUI (
       id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
       author = userRepository . getUserById ( author ) . toFrontReference () ,
       message = message ,
       files = files ?. let { files ->
         fileRepository . getHeadsByIds ( files )
           . map { it . toFrontReference () }
} ?: listOf () ,
       forwardFrom = forwardFrom ?. toFrontModel () ,
       replyTo = replyTo ?. toFrontModel ()
)
}

Tudo está muito claro, e isso é uma grande vantagem.

Utilizamos o processamento em lote e recebemos dados de um serviço externo em lotes. Mas o que está acontecendo com o desempenho?

Para cada mensagem, uma chamada para UserRemoteApi será feita para obter dados no campo autor e uma chamada para FileRemoteApi para receber todos os arquivos anexados. Parece ser tudo. Suponha que os campos forwardFrom e replyTo do ChatMessage sejam obtidos para que isso não exija chamadas extras. Mas transformá-los em ChatMessageUI levará à recursão, ou seja, o desempenho das contagens de chamadas pode aumentar bastante. Como observamos anteriormente, digamos que não temos muitos aninhamentos e a cadeia é limitada a três mensagens.

Como resultado, recebemos de duas a seis chamadas para serviços externos por mensagem e uma chamada JPA para todo o pacote de mensagens. O número total de chamadas varia de 2 * N + 1 a 6 * N + 1. Quanto é isso em unidades reais? Suponha que você precise de 20 postagens para renderizar uma página. Para obtê-los, você precisa de 4 sa 10 s. Horrível Eu gostaria de conhecer os 500 ms. E como o front-end sonhava em fazer uma rolagem contínua, os requisitos de desempenho desse endpoint podem ser dobrados.

Prós:

1. O código é conciso e auto-documentado (o sonho do suporte).
2. O código é simples, então quase não há oportunidades para disparar na perna.
3. O processamento em lote não parece estranho e organicamente se encaixa na lógica.
4. As alterações lógicas serão feitas facilmente e serão locais.

Menos:

Terrível desempenho devido ao fato de os pacotes serem muito pequenos.

Essa abordagem geralmente pode ser vista em serviços simples ou em protótipos. Se a velocidade da mudança é importante, dificilmente vale a pena complicar o sistema. Ao mesmo tempo, para nosso serviço muito simples, o desempenho é terrível, portanto o escopo de aplicabilidade dessa abordagem é muito restrito.

Processamento paralelo ingênuo

Você pode começar a processar todas as mensagens em paralelo - isso eliminará um aumento linear no tempo, dependendo do número de mensagens. Essa não é uma maneira particularmente boa, pois levará a um grande pico de carga no serviço externo.

A implementação do processamento paralelo é muito simples:

override fun getLast ( n : Int ) =
   messageRepository . findLast ( n ) . parallelStream ()
     . map { it . toFrontModel () }
     . collect ( toList ())

Usando o processamento paralelo de mensagens, obtemos de 300 a 700 ms idealmente, o que é muito melhor do que com uma implementação ingênua, mas ainda não é rápida o suficiente.

Com essa abordagem, as solicitações para userRepository e fileRepository serão executadas de forma síncrona, o que não é muito eficiente. Para corrigir isso, você precisará alterar bastante a lógica das chamadas. Por exemplo, através do CompletionStage (aka CompletableFuture):

private fun ChatMessage . toFrontModel () : ChatMessageUI =
   CompletableFuture . supplyAsync {
     userRepository . getUserById ( author ) . toFrontReference ()
   } . thenCombine (
     files ?. let {
       CompletableFuture . supplyAsync {
         fileRepository . getHeadsByIds ( files ) . map { it . toFrontReference () }
}
} ?: CompletableFuture . completedFuture ( listOf ())
) { author , files ->
     ChatMessageUI (
       id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
       author = author ,
       message = message ,
       files = files ,
       forwardFrom = forwardFrom ?. toFrontModel () ,
       replyTo = replyTo ?. toFrontModel ()
)
   } . get () !!

Pode-se ver que o código de mapeamento inicialmente simples se tornou menos claro. Isso ocorre porque tivemos que separar chamadas de serviço externas de onde os resultados foram usados. Isso por si só não é ruim. Mas a combinação de chamadas não parece muito elegante e se assemelha a um típico "macarrão" reativo.

Se você usar corotinas, tudo parecerá mais decente:

private fun ChatMessage . toFrontModel () : ChatMessageUI =
   join (
     { userRepository . getUserById ( author ) . toFrontReference () } ,
     { files ?. let { fileRepository . getHeadsByIds ( files )
       . map { it . toFrontReference () } } ?: listOf () }
   ) . let { ( author , files ) ->
     ChatMessageUI (
       id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
       author = author ,
       message = message ,
       files = files ,
       forwardFrom = forwardFrom ?. toFrontModel () ,
       replyTo = replyTo ?. toFrontModel ()
)
   }

Onde:

fun < A , B > join ( a : () -> A , b : () -> B ) =
   runBlocking ( IO ) {
     awaitAll ( async { a () } , async { b () } )
   } . let {
     it [ 0 ] as A to it [ 1 ] as B
   }

Teoricamente, usando esse processamento paralelo, obtemos 200 a 400 ms, o que já está próximo de nossas expectativas.

Infelizmente, essa boa paralelização não acontece, e a recompensa é bastante cruel: com apenas alguns usuários trabalhando ao mesmo tempo, uma enxurrada de solicitações cairá sobre os serviços, que ainda não serão processados ​​em paralelo, portanto, retornaremos aos nossos tristes 4 s.

Meu resultado ao usar esse serviço é 1300-1700 ms para processar 20 mensagens. Isso é mais rápido do que na primeira implementação, mas ainda não resolve o problema.


Prós:

1. Fácil implementação do processamento simultâneo de mensagens.
2. Boa escalabilidade.

Contras:

1. A necessidade de separar o recebimento de dados do processamento em solicitações de processamento paralelo para diferentes serviços.
2. Maior carga em serviços de terceiros.

Pode-se ver que o escopo de aplicabilidade é aproximadamente o mesmo da abordagem ingênua. O uso do método de consulta paralela faz sentido se você deseja aumentar o desempenho do seu serviço várias vezes devido à exploração impiedosa de outras pessoas. No nosso exemplo, a produtividade aumentou 2,5 vezes, mas isso claramente não é suficiente.

Armazenamento em cache

Você pode fazer o armazenamento em cache no estilo JPA para serviços externos, ou seja, armazenar objetos recebidos dentro da sessão para não recebê-los novamente (inclusive durante o processamento em lote). Você pode fazer esses caches você mesmo, pode usar o Spring com seu @Cacheable, além de sempre usar um cache pronto como o EhCache manualmente.

O problema geral estará relacionado ao fato de que há bom senso nos caches apenas se houver ocorrências. No nosso caso, as ocorrências no campo do autor (por exemplo, 50%) são muito prováveis ​​e não haverá ocorrências nos arquivos. Essa abordagem trará algumas melhorias, mas o desempenho não mudará radicalmente (e precisamos de uma inovação).

Caches entre sessões (longos) requerem lógica de invalidação complexa. Em geral, quanto mais tarde você chegar ao ponto de resolver problemas de desempenho com caches intersessionais, melhor.

Prós:

1. Implemente o cache sem alterar o código.
2. O desempenho aumenta várias vezes (em alguns casos).

Contras:

1. Possibilidade de degradação do desempenho se usado incorretamente.
2. Sobrecarga de memória grande, especialmente com caches longos.
3. Invalidação complexa, erros nos quais levarão a problemas difíceis em tempo de execução.

Muitas vezes, os caches são usados ​​apenas para corrigir rapidamente problemas de design. Isso não significa que eles não precisam ser usados. No entanto, sempre vale a pena tratá-los com cautela e primeiro avaliar o ganho de desempenho resultante e somente então tomar uma decisão.

No nosso exemplo, os caches terão um aumento de desempenho de cerca de 25%. Ao mesmo tempo, os caches têm muitas desvantagens, então eu não os usaria aqui.

Sumário

Portanto, analisamos a implementação ingênua de um serviço que usa o processamento em lote e algumas maneiras simples de acelerá-lo.

A principal vantagem de todos esses métodos é a simplicidade, da qual existem muitas consequências agradáveis.

Um problema comum com esses métodos é o desempenho ruim, principalmente devido ao tamanho do pacote. Portanto, se essas soluções não lhe agradam, vale a pena considerar métodos mais radicais.

Existem duas áreas principais nas quais você pode procurar soluções:

• Trabalho assíncrono com dados (requer uma mudança de paradigma, portanto, este artigo não é considerado)
• ampliação de pacotes, mantendo o processamento síncrono.

O aumento dos pacotes reduzirá bastante o número de chamadas externas e, ao mesmo tempo, manterá o código síncrono. A próxima parte do artigo será dedicada a este tópico.