Substituindo o RabbitMQ pelo MongoDB

Em abril, nós dispensamos o RabbitMQ em favor da construção do nosso próprio sistema de queuing usando o MongoDB. Isso está em produção desde então e tem funcionado muito bem.

Em setembro, eu fiz uma apresentação na conferência do MongoUK e no encontro London DevOps entrando em detalhes mais técnicos. A apresentação está inserida abaixo junto com um resumo completo. 

MongoUK 2011 – Rplacing RabbitMQ with MongoDB

View more presentations from Boxed Ice

Por que usar um queue?

• Processamento de background – tarefas que não precisam ser realizadas imediatamente enquanto o usuário espera, ou até tarefas pelas quais o usuário terá que esperar, mas que terão que ser executadas por processos separados. A chave aqui é executar as tarefas assincronamente, de forma que o usuário esteja ou fazendo outra coisa ou esperando com uma página de status.
• Exemplos desses tipos de trabalhos poderiam enviar notificações por e-mail, em que um atraso pequeno é crítico ou quando existem muito itens a serem enviados, por exemplo e-mail de massa.
• Comunicação interna – streaming de eventos através de sua infraestrutura para diferentes nós ou componentes.

Do que você precisa em um sistema de queueing?

• Consumidores – eles escutam a queue e agarram a mensagem a medida que ela chega, para executar as instruções contidas dentro dela. Geralmente são processos idênticos para que várias mensagens sejam processadas ao mesmo tempo.
• Atômico – você não quer vários consumidores puxando a mesma mensagem e causando duplicação. Então quando um consumidor compreende uma mensagem, nenhum outro consumidor deveria ser capaz de processá-la.
• Velocidade – os itens devem ser inseridos dentro da queue rapidamente. Da mesma maneira, eles devem ser retirados rapidamente. O tempo real de processamento pode variar dependamente do que está sendo feito, mas também das necessidades da queue de lidar com grandes números de leituras/escritas.
• Coleção de lixo (GC) – se um consumidor morre no meio do processo, então a mensagem que ele estava segurando deve ser retornada para a queue para outro consumidor pegar.

Implementação – Consumidores

A interação com o RabbitMQ é feita através do AMQP, que é um protocolo especificamente designado para sistemas de queue. Existem várias bibliotecas para diferentes linguagens que implementam isso, e apesar de não termos tido problemas na utilização da biblioteca Python AMQP, as bibliotecas PHP que testamos quebraram com altos carregamentos (apesar de eu saber que houve melhorias na biblioteca PHP desde então). O MongoDB tem bibliotecas similares disponíveis para muitas linguagens diferentes, e todas utilizam o Protocolo Mongo Wire, que temos utilizado para carregamentos maiores por muitos anos.

Ao mudarmos da biblioteca Python AMQP (pika) para o driver Python Mongo (pymongo), nós vimos uma redução significante nas despesas gerais, o que nos levou a concluir que poderíamos rodar mais consumidores por servidor e reduzir o número de servidores que precisávamos para processar o mesmo número de mensagens que chegavam.

O requerimento atômico é lidado pelo AMQP usando os métodos consume/ack e no MongoDB você usa o comando findAndModify. A forma geral é:

db.runCommand( { findAndModify : collection, { options } } )<br>

Ele irá retornar o primeiro documento que encontrar que corresponda às suas opcões, então as opcões em que estamos interessados são:

• where – esta consulta será executada para retornar um único documento. Para implementar a queuing, precisamos de vários meta campos como parte de cada método. Usamos inProg e done para que possamos manter um registro e consultas contra processos pendentes:

{'where' { 'inProg': false, 'done': false } }<br>

• sort – como findAndModify retorna o primeiro documento que ele encontra, você pode querer classificar os resultados para que você possa retornar o documento mais antigo (p.e. classificando por um campo de timestamp ou por _id, o que inclui o timestamp como parte do ID). Você também poderia configurar um campo de prioridade e dar às mensagens diferentes níveis de importância para aquelas que você quer processar primeiro.
• update – nós fazemos uma query para campos específicos, mas queremos também atualizá-los para que a query continue a funcionar (p.e. contra mensagens pendentes). Então atualizamos o campo inProg e também configuramos um (usado no processo GC depois).

{'update' : {'$set' : { 'inProg' : true, 't' : new Date() } } }  

Não estamos preocupados com os outros campos para o objetivo do queuing.

A query findAndModify precisa ser executada periodicamente para puxar as mensagens do banco de dados. Como fazer isso irá depender da linguagem que você estiver usando, mas nós utilizamos Python que pode fazer uso do módulo sched:

1 def run(self):<br>2    import time, sched<br>3    sc = sched.scheduler(time.time, time.sleep)<br>4    self.poll(sc)<br>5    sc.run()<br>6  <br>7 def poll(self, sc):<br>8    doStuff()<br>9    sc.enter(10, 1, self.poll, (sc,))

Aqui, o método poll será chamado a cada 10 segundos e irá chamar o método doStuff e então reagendar a próxima execução. O doStuff é onde você colocaria seu chamado para findAndModify, além de qualquer processo que você queira fazer com o documento retornado.

Mantemos esses scripts rodando como daemons, usando nossa classe daemon, que está disponível de graça sob uma licença open source.

Implementação – GC

Como parte do nosso chamado para findAndModify, configuramos um timestamp para que possamos saber quando a mensagem foi retirada. Baseado em quanto tempo esperamos que os consumidores demorem por mensagem, podemos consultar facilmente os documentos que estão marcados como em progresso mas que o timestamp foi de um tempo atrás.

Por exemplo, se esperamos que os itens sejam processados dentro de 10 segundos:

1 now = datetime.datetime.now()<br>2 difference = datetime.timedelta(seconds=10)<br>3 timeout = now - difference<br>4  <br>5 queue.find({'inProg' : True, 'start' : {'$lte' : timeout} })

Podemos então atualizar esses documentos para resetar a flag inProg, retornando-os efetivamente para a queue para outro consumidor trabalhar.

Outras considerações

• Tolerância a falhas – conjuntos de réplicas no MongoDB funcionam extremamente bem para permitir redundância e failover automáticos. O motivo original pelo qual removemos o RabbitMQ foi a falta de suporte embutido para redundância através de nós/datacenters múltiplos. Isso agora é possível com o RabbitMQ 2.6.0.
• Consistência – journaling é habilitado por padrão no MongoDB 2.0, o que significa que os dados são enviados para arquivo journal a cada 100ms (por padrão) e então para os arquivos de dados a cada 60 segundos (por padrão). Uma vez que os dados estão no diário, eles podem ser considerados seguros. Isso deve ser utilizado em combinação com os conjuntos de réplicas para assegurar que os dados vão para múltiplos locais. Você pode configurar flags em suas queries para assegurar que os dados tenham sido escritos para o journal e/ou enviados para o disco, além de terem sido escritos em nós replicados se você precisar se certificar de que não perderá nada na queue.
• Escalada – usamos uma coleção capped para nossas queues. Elas são criadas com tamanho fixo e os documentos antigos são descartados na mesma ordem em que foram inseridos, de modo que não precisamos nos preocupar em remover documentos, o que pode ser uma operação bastante intensa. No entanto, em certo ponto, você irá atingir em um limite em que a natureza atômica dos travamentos do findAndModify significam que você precisa escalar suas escritas através de várias máquinas. Isso deve ser feito usando o sharding de coleções regulares, uma vez que coleções capped não podem ser compartilhadas.

?

Texto original disponível em http://blog.boxedice.com/2011/09/28/replacing-rabbitmq-with-mongodb/