Olá pessoal. Meu nome é Alexander, sou desenvolvedor Java no grupo de empresas Tinkoff.

Neste artigo, quero compartilhar minha experiência na solução de problemas associados à sincronização do estado de caches em sistemas distribuídos. Nós os encontramos, dividindo nosso aplicativo monolítico em microsserviços . Obviamente, falaremos sobre o armazenamento em cache de dados no nível da JVM, porque com os caches externos, os problemas de sincronização são resolvidos fora do contexto do aplicativo.

Neste artigo, falarei sobre nossa experiência de mudar para uma arquitetura orientada a serviços, acompanhada de uma mudança para o Kubernetes, e sobre a solução de problemas relacionados. Consideraremos a abordagem para organizar o sistema de cache distribuído In-Memory Data Grid (IMDG), suas vantagens e desvantagens, por causa do qual decidimos escrever nossa própria solução.

Este artigo discute um projeto cujo back-end é escrito em Java. Portanto, também falaremos sobre padrões no campo de cache temporário na memória. Discutimos a especificação JSR-107, a especificação JSR-347 com falha e os recursos de armazenamento em cache no Spring. Bem-vindo ao gato!

E vamos cortar o aplicativo em serviços ...

Vamos seguir para a arquitetura orientada a serviços e para o Kubernetes - foi o que decidimos há pouco mais de 6 meses. Por um longo tempo, nosso projeto foi um monólito, muitos problemas relacionados à dívida técnica acumulada e escrevemos novos módulos de aplicativos como serviços separados. Como resultado, a transição para uma arquitetura orientada a serviços e um corte de monólito era inevitável.

Nosso aplicativo é carregado, em média 500 rps chega a serviços da web (no pico atinge 900 rps). Para coletar todo o modelo de dados em resposta a cada solicitação, é necessário ir aos vários caches várias centenas de vezes.

Tentamos acessar o cache remoto no máximo três vezes por solicitação, dependendo do conjunto de dados necessário, e nos caches internos da JVM, a carga atinge 90.000 rps por cache. Temos cerca de 30 desses caches para uma variedade de entidades e o DTO-shki. Em alguns caches carregados, não podemos nem mesmo excluir o valor, pois isso pode levar a um aumento no tempo de resposta dos serviços da web e a uma falha no aplicativo.


É assim que parece o monitoramento de carga, removido dos caches internos em cada nó durante o dia. De acordo com o perfil de carregamento, é fácil ver que a maioria das solicitações são dados de leitura. Um carregamento de gravação uniforme é devido à atualização de valores em caches em uma determinada frequência.

O tempo de inatividade não é válido para nosso aplicativo. Portanto, com a finalidade de uma implantação contínua, sempre equilibramos todo o tráfego recebido em dois nós e implantamos o aplicativo usando o método Rolling Update. O Kubernetes se tornou nossa solução ideal de infraestrutura ao mudar para serviços. Assim, resolvemos vários problemas ao mesmo tempo.

O problema de solicitar e configurar constantemente a infraestrutura para novos serviços

Nos foi dado um espaço para nome no cluster para cada circuito, que temos três: dev - para desenvolvedores, qa - para testadores, prod - para clientes.

Com o espaço para nome realçado, a adição de um novo serviço ou aplicativo se resume a escrever quatro manifestos: Implantação, Serviço, Ingress e ConfigMap.

Alta tolerância de carga

Os negócios estão em expansão e em constante crescimento - há um ano, a carga média era duas vezes menor que a atual.

A escala horizontal no Kubernetes permite nivelar as economias de escala com o aumento da carga de trabalho do projeto desenvolvido.

Manutenção, coleta e monitoramento de logs

A vida se torna muito mais fácil quando não há necessidade de adicionar logs ao sistema de registro ao adicionar cada nó, configurar a cerca de métricas (a menos que você tenha um sistema de monitoramento por push), executar configurações de rede e simplesmente instalar o software necessário para a operação.

Obviamente, tudo isso pode ser automatizado usando o Ansible ou o Terraform, mas no final, escrever vários manifestos para cada serviço é muito mais fácil.

Alta confiabilidade

O mecanismo interno do k8s de amostras de vitalidade e prontidão permite que você não se preocupe com o fato de o aplicativo começar a ficar mais lento ou parar completamente de responder.

O Kubernetes agora controla o ciclo de vida dos pods de lareira que contêm contêineres de aplicativos e o tráfego direcionado a eles.

Juntamente com as comodidades descritas, precisamos resolver vários problemas para tornar os serviços adequados para a escala horizontal e o uso de um modelo de dados comum para muitos serviços. Foi necessário resolver dois problemas:

1. O estado do aplicativo. Quando o projeto é implantado no cluster k8s, os pods com contêineres da nova versão do aplicativo começam a ser criados que não estão relacionados ao estado dos pods da versão anterior. Novos pods de aplicativos podem ser criados em servidores de cluster arbitrários que atendem às restrições especificadas. Além disso, agora todos os contêineres de aplicativos em execução no pod do Kubernetes podem ser destruídos a qualquer momento, se o probe Liveness disser que precisa ser reiniciado.
2. Consistência de dados. É necessário manter a consistência e a integridade dos dados entre si em todos os nós. Isso é especialmente verdadeiro se vários nós funcionarem em um único modelo de dados. É inaceitável que, quando solicitações para diferentes nós do aplicativo na resposta, dados inconsistentes cheguem ao cliente.

No desenvolvimento moderno de sistemas escaláveis, a arquitetura Stateless é a solução para os problemas acima. Nós nos livramos do primeiro problema movendo todas as estatísticas para o armazenamento em nuvem S3.

No entanto, devido à necessidade de agregar um modelo de dados complexo e economizar tempo de resposta de nossos serviços da web, não podemos nos recusar a armazenar dados em caches na memória. Para resolver o segundo problema, eles criaram uma biblioteca para sincronizar o estado dos caches internos de nós individuais.

Sincronizamos caches em nós separados

Como dados iniciais, temos um sistema distribuído composto por N nós. Cada nó possui cerca de 20 caches na memória, cujos dados são atualizados várias vezes por hora.

A maioria dos caches possui uma política de atualização de dados TTL (tempo de vida útil); alguns dados são atualizados com uma operação CRON a cada 20 minutos devido ao alto carregamento. A carga de trabalho nos caches varia de vários milhares de rps à noite a várias dezenas de milhares durante o dia. A carga de pico, como regra, não excede 100.000 rps. O número de registros no armazenamento temporário não excede várias centenas de milhares e é colocado no heap de um nó.

Nossa tarefa é obter consistência de dados entre o mesmo cache em nós diferentes, bem como o menor tempo de resposta possível. Considere o que geralmente há maneiras de resolver esse problema.

A primeira e mais simples solução que vem à mente é colocar todas as informações em um cache remoto. Nesse caso, você pode se livrar completamente do estado do aplicativo, não pensar nos problemas de obter consistência e ter um único ponto de acesso a um data warehouse temporário.


Esse método de armazenamento temporário de dados é bastante simples, e nós o usamos. Armazenamos em cache parte dos dados no Redis , que é um armazenamento de dados NoSQL na RAM. No Redis, geralmente registramos uma estrutura de resposta de serviço da Web e, para cada solicitação, precisamos enriquecer esses dados com informações relevantes, para as quais precisamos enviar várias centenas de solicitações ao cache local.

Obviamente, não podemos retirar os dados dos caches internos para armazenamento remoto, pois o custo de transmissão de um volume de tráfego pela rede não nos permitirá atingir o tempo de resposta necessário.

A segunda opção é usar um IMDG ( In-Memory Data Grid ), que é um cache distribuído na memória. O esquema dessa solução é o seguinte:


A arquitetura IMDG é baseada no princípio de Particionamento de Dados de caches internos de nós individuais. De fato, isso pode ser chamado de tabela de hash distribuída em um cluster de nós. O IMDG é considerado uma das implementações mais rápidas de armazenamento distribuído temporário.

Existem muitas implementações do IMDG, sendo a mais popular a Hazelcast . O cache distribuído permite armazenar dados na RAM em vários nós de aplicativos com um nível aceitável de confiabilidade e preservação da consistência, o que é alcançado pela replicação de dados.

A tarefa de construir um cache distribuído não é fácil, no entanto, o uso de uma solução IMDG pronta para nós pode ser um bom substituto para os caches da JVM e eliminar os problemas de replicação, consistência e distribuição de dados entre todos os nós de aplicativos.

A maioria dos fornecedores de IMDG para aplicativos Java implementa o JSR-107 , a API Java padrão para trabalhar com caches internos. Em geral, esse padrão tem uma história bastante grande, que discutirei em mais detalhes abaixo.

Era uma vez idéias para implementar sua interface para interagir com o IMDG - JSR 347 . Mas a implementação dessa API não recebeu suporte suficiente da comunidade Java e agora temos uma interface única para interagir com os caches na memória, independentemente da arquitetura do nosso aplicativo. Bom ou ruim é outra questão, mas nos permite ignorar completamente todas as dificuldades de implementar um cache distribuído na memória e trabalhar com ele como um cache de um aplicativo monolítico.

Apesar das vantagens óbvias do uso do IMDG, essa solução ainda é mais lenta que o cache da JVM padrão, devido à sobrecarga de garantir a replicação contínua dos dados distribuídos entre vários nós da JVM, além de fazer o backup desses dados. No nosso caso, a quantidade de dados para armazenamento temporário não era tão grande, os dados com margem cabiam na memória de um aplicativo; portanto, sua alocação em várias JVMs parecia uma solução desnecessária. E o tráfego de rede adicional entre nós de aplicativos sob cargas pesadas pode afetar significativamente o desempenho e aumentar o tempo de resposta dos serviços da web. No final, decidimos escrever nossa própria solução para esse problema.

Deixamos os caches na memória como um armazenamento temporário de dados e, para manter a consistência, usamos o gerenciador de filas RabbitMQ. Adotamos o padrão de design comportamental “Publisher - Subscriber” e mantivemos a relevância dos dados excluindo o registro modificado do cache de cada nó. O esquema da solução é o seguinte:


O diagrama mostra um cluster de N nós, cada um com um cache de memória padrão. Todos os nós usam um modelo de dados comum e devem ser consistentes. No primeiro acesso ao cache por uma chave arbitrária, o valor no cache está ausente e colocamos o valor real do banco de dados nele. Com qualquer alteração - exclua o registro.

As informações reais na resposta do cache aqui são fornecidas pela sincronização da exclusão de uma entrada quando ela é alterada em qualquer um dos nós. Cada nó no sistema possui uma fila no gerenciador de filas RabbitMQ. A gravação em todas as filas é feita através de um ponto de acesso comum do tipo Tópico. Isso significa que as mensagens enviadas ao Tópico se enquadram em todas as filas associadas a ele. Portanto, ao alterar o valor em qualquer nó do sistema, esse valor será excluído do armazenamento temporário de cada nó e o acesso subsequente iniciará a gravação do valor atual no cache do banco de dados.

A propósito, um mecanismo PUB / SUB semelhante existe no Redis. Mas, na minha opinião, ainda é melhor usar o gerenciador de filas para trabalhar com filas, e o RabbitMQ foi perfeito para nossa tarefa.

Norma JSR 107 e sua implementação

A API Java Cache padrão para armazenamento temporário de dados na memória (especificação JSR-107 ) tem um histórico bastante longo, desenvolvida por 12 anos.

Durante tanto tempo, as abordagens para o desenvolvimento de software mudaram, os monólitos foram substituídos pela arquitetura de microsserviços. Devido a uma falta tão longa de especificações para a API de cache, houve até pedidos para desenvolver caches de API para sistemas distribuídos JSR-347 (grades de dados para a plataforma Java). Porém, após o tão esperado lançamento do JSR-107 e o lançamento do JCache, a solicitação para criar uma especificação separada para sistemas distribuídos foi retirada.

Nos últimos 12 anos no mercado, o local para armazenamento temporário de dados mudou de HashMap para ConcurrentHashMap com o lançamento do Java 1.5 e, mais tarde, surgiram muitas implementações de código aberto do cache na memória.

Após o lançamento do JSR-107, as soluções dos fornecedores começaram a implementar gradualmente a nova especificação. Para o JCache, existem até fornecedores especializados em cache distribuído - os próprios Data Grids, cuja especificação nunca foi implementada.

Considere o que consiste o pacote javax.cache e como obter uma instância de cache para nosso aplicativo:

CachingProvider provider = Caching.getCachingProvider("org.cache2k.jcache.provider.JCacheProvider"); CacheManager cacheManager = provider.getCacheManager(); CacheConfiguration<Integer, String> config = new MutableConfiguration<Integer, String>() .setTypes(Integer.class, String.class) .setReadThrough(true) . . .; Cache<Integer, String> cache = cacheManager.createCache(cacheName, config); 

Aqui, o cache é um carregador de inicialização para o CachingProvider.

No nosso caso, o JCacheProvider, que é a implementação cache2k do SPI do provedor JSR-107, será carregado do ClassLoader. Para o carregador, talvez você não precise especificar a implementação do provedor, mas ele tentará carregar a implementação que

META-INF / services / javax.cache.spi.CachingProvider

De qualquer forma, no ClassLoader, deve haver uma única implementação CachingProvider.

Se você usar a biblioteca javax.cache sem nenhuma implementação, uma exceção será lançada ao tentar criar o JCache. O objetivo do provedor é criar e gerenciar o ciclo de vida do CacheManager, que, por sua vez, é responsável por gerenciar e configurar os caches. Portanto, para criar um cache, você deve seguir o seguinte caminho:


Os caches padrão criados usando o CacheManager devem ter uma configuração compatível com a implementação. O CacheConfiguration padrão com parâmetros fornecido pelo javax.cache pode ser estendido para uma implementação específica do CacheProvider.

Hoje, existem dezenas de implementações diferentes da especificação JSR-107: Ehcache , Goiaba , cafeína , cache2k . Muitas implementações são In-Memory Data Grid em sistemas distribuídos - Hazelcast , Oracle Coherence .

Também há muitas implementações de armazenamento temporário que não oferecem suporte à API padrão. Por um longo tempo em nosso projeto, usamos o Ehcache 2, que não é compatível com o JCache (a implementação da especificação apareceu no Ehcache 3). A necessidade de uma transição para uma implementação compatível com JCache apareceu com a necessidade de monitorar o status dos caches na memória. Usando o MetricRegistry padrão, foi possível fixar o monitoramento apenas com a ajuda da implementação JCacheGaugeSet, que coleta métricas do JCache padrão.

Como escolher a implementação de cache na memória apropriada para o seu projeto? Talvez você deva prestar atenção ao seguinte:

1. Você precisa de suporte para a especificação JSR-107.
2. Também vale a pena prestar atenção na velocidade da implementação selecionada. Sob cargas pesadas, o desempenho dos caches internos pode ter um impacto significativo no tempo de resposta do seu sistema.
3. Suporte na primavera. Se você usar a estrutura conhecida em seu projeto, vale a pena considerar o fato de que nem toda implementação de cache da JVM possui um CacheManager compatível no Spring.

Se você estiver usando ativamente o Spring em seu projeto, assim como nós, então para o cache de dados, provavelmente seguirá a abordagem orientada a aspectos (AOP) e usará a anotação @Cacheable. O Spring usa seu próprio CacheManager SPI para os aspectos funcionarem. O seguinte bean é necessário para que os caches de primavera funcionem:

 @Bean public org.springframework.cache.CacheManager cacheManager() { CachingProvider provider = Caching.getCachingProvider(); CacheManager cacheManager = provider.getCacheManager(); return new JCacheCacheManager(cacheManager); } 

Para trabalhar com caches no paradigma AOP, considerações transacionais também devem ser consideradas. O cache de primavera deve necessariamente suportar o gerenciamento de transações. Para isso, o Spring CacheManager herda as propriedades AbstractTransactionSupportingCacheManager, que podem ser usadas para sincronizar operações de put / evict executadas em uma transação e executá-las somente após o comprometimento de uma transação bem-sucedida.

O exemplo acima mostra o uso do wrapper JCacheCacheManager para o gerenciador de especificações de cache. Isso significa que qualquer implementação JSR-107 também tem compatibilidade com o Spring CacheManager. Esse é outro motivo para escolher um cache de memória com suporte para a especificação JSR do seu projeto. Mas se esse suporte ainda não for necessário, mas eu realmente quero usar o @Cacheable, você terá suporte para mais duas soluções de cache interno: EhCacheCacheManager e CaffeineCacheManager.

Ao escolher a implementação do cache na memória, não levamos em consideração o suporte do IMDG para sistemas distribuídos, como mencionado anteriormente. Para manter o desempenho dos caches da JVM em nosso sistema, criamos nossa própria solução.

Limpando caches em um sistema distribuído

IMDGs modernos usados ​​em projetos com arquitetura de microsserviços permitem distribuir dados na memória entre todos os nós em funcionamento do sistema usando o particionamento de dados escalável com o nível de redundância necessário.

Nesse caso, há muitos problemas associados à sincronização, consistência dos dados e assim por diante, sem mencionar o aumento no tempo de acesso ao armazenamento temporário. Esse esquema é redundante se a quantidade de dados usada se encaixar na RAM de um nó e, para manter a consistência dos dados, basta excluir essa entrada em todos os nós para qualquer alteração no valor do cache.

Ao implementar essa solução, vem à mente a idéia de usar algum EventListener, no JCache, existe um CacheEntryRemovedListener para o evento de excluir uma entrada do cache. Parece que é suficiente adicionar sua própria implementação do Ouvinte, que enviará mensagens para o tópico quando o registro for excluído, e o cache eutético em todos os nós estiver pronto - desde que cada nó escute eventos da fila associada ao tópico geral, conforme mostrado no diagrama acima.

Ao usar essa solução, os dados em nós diferentes serão inconsistentes devido ao fato de que EventLists em qualquer processo de implementação do JCache após o evento ocorrer. Ou seja, se não houver registro no cache local para a chave especificada e houver um registro para a mesma chave em qualquer outro nó, o evento não será enviado ao tópico.


Considere outras maneiras de capturar o evento de um valor ser excluído do cache local.

No pacote javax.cache.event, ao lado de EventListeners, também há um CacheEntryEventFilter, que, de acordo com o JavaDoc, é usado para verificar qualquer evento CacheEntryEvent antes de enviar esse evento ao CacheEntryListener, seja um registro, exclusão, atualização ou evento relacionado à expiração do registro. em cache. Ao usar o filtro, nosso problema permanecerá, pois a lógica será executada após o log do evento CacheEntryEvent e após a operação CRUD no cache.

No entanto, é possível capturar o início de um evento para excluir um registro do cache. Para fazer isso, use a ferramenta interna do JCache que permite usar as especificações da API para gravar e carregar dados de uma fonte externa, se eles não estiverem no cache. Existem duas interfaces para isso no pacote javax.cache.integration:

• CacheLoader - para carregar os dados solicitados pela chave, se não houver entradas no cache.
• CacheWriter - para usar a gravação, exclusão e atualização de dados em um recurso externo ao invocar as operações de cache correspondentes.

Para garantir consistência, os métodos CacheWriter são atômicos em relação à operação de cache correspondente. Parece que encontramos uma solução para o nosso problema.

Agora, podemos manter a consistência da resposta dos caches na memória nos nós ao usar nossa implementação do CacheWriter, que envia eventos para o tópico do RabbitMQ sempre que houver alguma alteração no registro no cache local.

Conclusão

Ao desenvolver qualquer projeto, ao procurar uma solução adequada para problemas emergentes, é preciso levar em consideração sua especificidade. No nosso caso, os recursos característicos do modelo de dados do projeto, o código herdado herdado e a natureza da carga não permitiram o uso de nenhuma das soluções existentes para o problema de armazenamento em cache distribuído.

É muito difícil tornar uma implementação universal aplicável a qualquer sistema desenvolvido. Para cada uma dessas implementações, existem condições ideais de uso. No nosso caso, as especificidades do projeto levaram à solução descrita neste artigo. Se alguém tiver um problema semelhante, ficaremos felizes em compartilhar nossa solução e publicá-la no GitHub.