🐇 Work/Task Queues

[lucasgianine]

Work Queues

Vamos trabalhar em criar Work Queues (ou Task Queues) para distribuir tarefas demoradas entre vários workers, ou seja, quando uma tarefa exije muitos recursos, todo fluxo espera que ela seja concluída para que a mensagem seja exibida, a ideia do Work Queues é que agendemos a tarefa para que ela seja feita mais tarde.

flowchart LR
  P["Producer"]
  Queue["Task"]
  C1["Consumer 1"]
  C2["Consumer 2"]

  P --> Queue --> C1
  Queue --> C2

Loading

Utilize esses comandos para teste:

  # shell 1
  npm run consumer

  # -> [*] Waiting for messages. To exit press CTRL+C
  # -> [x] Received <mesagem>!
  # -> [x] Done

  # shell 2
  npm run producer <mensagem>

  # -> [x] Sent: <mensagem>

Round-robin dispatching

Durante esse teste, existe uma maneira de distribuição que é chamada de round-robin, que por padrão, cada mensagem será enviada por sequencia, fazendo com que cada consumidor receba o mesmo número de mensagens, a diferença é que as mensagens serão distribuídas em shell diferentes.
Se executarmos o npm run consumer em dois shell, e depois enviarmos as mensagems, podemos ver que cada um terá diferentes resultados, pois foram distribuídos em sequeência.

Message acknowledgment

Ás vezes, fazer uma tarefa leva alguns segundos, e por sua vez, algumas tarefas acabam morrendo no meio do caminho sem ter chance de serem executadas, e tudo que estava sendo processado acaba sendo perdido no meio do caminho, e se fizessemos algo para que, assim que uma tarefa morrer, ela passasse seu trabalho (que estava em andamento) para o próximo producer?

  // worker.js
  noAck: false // acknowledgment mode

Aplicar noAck como false (que anteriormente era true) garante que, mesmo que você encerre um worker (com CTRL+C) nada será perdido, todas as mensagens não confirmadas serão reenviadas (desde que esteja no mesmo canal em que ela foi enviada).

Message durability

Ainda que garantimos que mesmo que o consumer acabe, as mensagens ainda existam, não garantimos que se o servidor RabbitMQ parar essas mesmas mensagens deixem de existir, pois quando o RabbitMQ para ele esquece todas as queues e mensagens, a não ser que façamos ele lembrar.
Fazemos ele se lembrar das queues utilizando durable, e para as mensagens persistent

  channel.assertQueue(queue, {
      durable: true
    })

  channel.sendToQueue(queue, Buffer.from(message), {
    persistent: true
  })

Fair dispatch

Esse método é interessante por um motivo: Supondo que mensagens ímpares sejam pesadas e mensagens pares sejam leves, um trabalhador ficará constantemente ocupado e outro quase não terá trabalho para fazer, o Rabbit faz com que as mensagens sejam enviadas uniformemente, sem se preocupar com isso, pois ele só despacha as mensagens que entram na queue/fila

flowchart LR
  P["Producer"]
  Queue["Task"]
  C1["Consumer 1"]
  C2["Consumer 2"]

  P --> Queue -- Prefetch=1 --> C1
  Queue -- Prefetch=1 --> C2

Loading

Pra corrigir esse feito, usamos prefetch com o valor 1 para que o Rabbit entenda não dará mais de uma mensagem para um trabalhador por vez enquanto a mensagem não for concluída até que um trabalhador esteja livre.

  channel.prefetch(1)