Uma novidade que vai deixar aqueles que ficavam comparando RabbitMQ e Kafka mais perdidos ainda. No dia 9 de julho saiu um o RabbitMQ 3.9.0 RC que conta com suporte a Streams.

A comparação direta era descabida até então: Enquanto o Kafka trabalha com streams o RabbitMQ lida com filas. Alguns comportamentos fundamentais nunca foram possíveis no Kafka, da mesma forma que alguns dos comportamentos do Kafka nunca foram possíveis com RabbitMQ.

Até agora!

Na versão 3.9 que saiu no último dia 9 de Julho, um recurso me chamou a atenção porque não teve alarde algum. Trata-se do suporte à streams. Agora o RabbitMQ trabalha tanto com Queues (chamadas de Classic Queues) quanto com Streams e isso é incrível.

As classic queues tem um comportamento mandatório de controle de processamento. Por isso em uma fila, uma mesma mensagem nunca é entregue para 2 consumidores.

Ou seja, se você tem 100 mensagens na fila, em condições perfeitas, com 2 consumidores, cada um receberia 50 mensagens, nenhuma mensagem é entregue 2 vezes ou para os 2 consumidores. Assim, até então, para que fazer com que 2 consumidores consumissem a mesma mensagem, eram necessárias 2 classic queues.

A regra é que se você quer entregar a mesma mensagem para diversos consumidores, cada consumidor precisa de uma fila.

A única exceção à essa regra diz respeito ao reprocessamento em caso de falha ou timeout.

Streams

As streams possuem consumo não destrutivo. Ou seja, consumir não apaga as mensagens.

Ao iniciar o consumo de mensagens (chamando o método model.BasicConsume) é necessário enviar um parâmetro OFFSET para determinar a partir de onde devemos começar a consumir a stream. Trata-se de um número, mas existem convenções como FIRST, LAST e NEXT, além do suporte a datas no padrão POSIX.

Exceto pela adição desse parâmetro (x-stream-offset) a coreografia de consumo segue a mesma das classic queues. Já na declaração da stream, precisamos adicionar o parâmetro “x-queue-type” com valor “stream” para que ele crie uma stream.

Diferente das filas, todos os consumidores de uma stream receberão todas as mensagens a partir do offset informado.

Embora esse comportamento só tenha chegado agora, após 14 anos desde seu lançamento em 2007, ele está no inconsciente e na expectativa de dá os primeiros passos com RabbitMQ. Seja pela ausência de preocupação nos produtores de tutoriais, ou pela falta de intimidade do leitor com documentações, ou mesmo pelo hábito e preconceito a respeito das Documentações que acaba não indo muito fundo no entendimento dos fundamentos da plataforma.

Esse novo recurso parece motivador, já que no caso de .NET temos um provider nativo de primeira linha. O nível de preocupação com nosso provider é alto e isso é bem interessante. Outra coisa interessante é que não é necessário um provider especial, o AMQP continua funcionando perfeitamente bem, inclusive com a coreografia de ack manual. Onde o ack tem o papel de alterar um offset default na stream.

A versão 3.9

Essa versão traz break changes na configuração, principalmente para quem usa Docker e Docker Compose para subir instâncias RabbitMQ.

Esses são detalhes para um novo post.

O que muda na API?

Criando Streams

Na declaração de uma Stream, usamos a mesma API que usávamos para declarar filas.

ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

Map<String, Object> arguments = new HashMap<>();
arguments.put("x-queue-type", "stream");
arguments.put("x-max-length-bytes", 20_000_000_000); // maximum stream size: 20 GB
arguments.put("x-stream-max-segment-size-bytes", 100_000_000); // size of segment files: 100 MB
channel.queueDeclare(
  "my-stream",
  true,         // durable
  false, false, // not exclusive, not auto-delete
  arguments
);

O AMQP sempre disponibilizou esses dicionário de argumentos adicionais, que na maioria das vezes não usávamos em muitos casos. Especificamente na criação da Queue, era um cenário comum. No entanto não precisávamos expressar o x-queue-type quando tratávamos de filas clássicas. Para declarar uma stream, você precisa informar o x-queue-type com o valor stream.

Consumo

No consumo a diferença segue o mesmo estilo, apenas 1 parâmetro novo: x-stream-offset.

channel.basicQos(100); // QoS must be specified
channel.basicConsume(
  "my-stream",
  false,
  Collections.singletonMap("x-stream-offset", "first"), // offset value
  (consumerTag, message) -> {
    // message processing
    // ...
   channel.basicAck(message.getEnvelope().getDeliveryTag(), false); // ack is required
  },
  consumerTag -> { });

A parte curiosa são as possibilidades de valores para x-stream-offset.

Convenções

"x-stream-offset", "first"

Valor

"x-stream-offset", 5000

Datas/Tempo

Date timestamp = new Date(System.currentTimeMillis() - 60 * 60 * 1_000);

"x-stream-offset", timestamp

Os exemplos de código foram tirados da documentação.

Conclusão

Me chama a atenção essa novidade, visto que já havia sido dito no post Understanding the Differences Between RabbitMQ vs Kafka em Nov/2020 que stream estava para chegar. Mas somente agora em Jul/2021, quase 1 ano depois, de fato temos a RC com streams chega ao github.

Para o meu uso, nada muda, os recursos mais importantes para mim são:

• Garantia de entrega
• Garantia de entrega única entre consumidores

Eu uso esses recursos para distribuir processamento entre consumidores e garantir que executarão a tarefa apesar de qualquer adversidade de infra, rede, indisponibilidade e downtime próprio ou de dependências. Óbvio que vou explorar a nova funcionalidade, Stream precisa entrar no meu curso de RabbitMQ para Aplicações .NET, mas embora possa fazer muita diferença na vida de muita gente, na minha por hora, não fará.