ApsaraMQ for RocketMQ:Filtragem de mensagens

Regras de tags

• Cada mensagem pode ter apenas uma tag.

• A tag é uma string sensível a maiúsculas e minúsculas. Por exemplo, TagA e tagA são tags diferentes.

• Mantenha as tags com menos de 128 caracteres.

Sintaxe da expressão de filtro

Caso de uso: roteamento de transações de e-commerce

Uma plataforma de e-commerce envia mensagens de pedidos, pagamentos e logística para um único tópico Trade_Topic. Diferentes sistemas downstream assinam com filtros de tags distintos:

Código de exemplo

Defina uma tag no lado do produtor:

Message message = messageBuilder.setTopic("topic")
                .setKeys("messageKey")
                .setTag("TagA")
                .setBody("messageBody".getBytes())
                .build();

Assine mensagens com uma única tag:

String topic = "Your Topic";
FilterExpression filterExpression = new FilterExpression("TagA", FilterExpressionType.TAG);
pushConsumer.subscribe(topic, filterExpression);

Assine mensagens correspondentes a qualquer uma das múltiplas tags:

String topic = "Your Topic";
FilterExpression filterExpression = new FilterExpression("TagA||TagB||TagC", FilterExpressionType.TAG);
pushConsumer.subscribe(topic, filterExpression);

Assine todas as mensagens do tópico:

String topic = "Your Topic";
FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
pushConsumer.subscribe(topic, filterExpression);

Regras de atributos

• Cada mensagem pode ter vários atributos personalizados. Cada atributo é um par chave-valor.

• As chaves de atributo devem começar com uma letra ou sublinhado (_) e conter apenas letras, dígitos e sublinhados.

Referência de sintaxe SQL-92

Escreva expressões de filtro usando os seguintes operadores SQL-92:

Restrições de valores numéricos:

• Os valores numéricos são inteiros de 32 bits, variando de -2.147.483.648 a 2.147.483.647.

• Strings conversíveis em números são tratadas como valores numéricos.

Como o broker lida com casos extremos

Se não for possível avaliar uma expressão de filtro corretamente, o broker descarta a mensagem em vez de entregá-la. Isso se aplica a três situações:

Caso de uso: filtragem de mensagens de logística por região

Uma plataforma de e-commerce envia mensagens de pedidos e logística para Trade_Topic. As mensagens de logística carregam um atributo Region com valores como Hangzhou ou Shanghai.

Código de exemplo

Defina uma tag e atributos personalizados no lado do produtor:

Message message = messageBuilder.setTopic("topic")
                .setKeys("messageKey")
                .setTag("messageTag")
                // Custom attribute: region
                .addProperty("Region", "Hangzhou")
                .setBody("messageBody".getBytes())
                .build();

Assine mensagens correspondentes a um único atributo:

String topic = "topic";
FilterExpression filterExpression = new FilterExpression(
    "Region IS NOT NULL AND Region='Hangzhou'", FilterExpressionType.SQL92);
simpleConsumer.subscribe(topic, filterExpression);

Assine mensagens correspondentes a múltiplos atributos:

String topic = "topic";
FilterExpression filterExpression = new FilterExpression(
    "Region IS NOT NULL AND price IS NOT NULL AND Region = 'Hangzhou' AND price > 30",
    FilterExpressionType.SQL92);
simpleConsumer.subscribe(topic, filterExpression);

Assine todas as mensagens do tópico:

String topic = "topic";
FilterExpression filterExpression = new FilterExpression("True", FilterExpressionType.SQL92);
simpleConsumer.subscribe(topic, filterExpression);

Por que consumidores no mesmo grupo perdem mensagens ao assinar tags diferentes?

Todos os consumidores em um grupo de consumidores devem usar a mesma expressão de filtro. Se o consumidor A assinar TagA enquanto o consumidor B assinar TagB, ocorrerá inconsistência de assinatura e haverá perda de mensagens. Alinhe as expressões de filtro em todos os consumidores do grupo ou divida-os em grupos de consumidores separados.

Como a contagem de mensagens consumidas é calculada quando a filtragem está ativa?

Apenas as mensagens que passam pela expressão de filtro contam para o consumo. As mensagens filtradas são excluídas da contagem.

Por que as mensagens se acumulam mesmo quando os consumidores mostram consumo zero?

As mensagens que não correspondem ao filtro ativo ainda existem no tópico e contam para o backlog. Esse comportamento é esperado. Consulte Cálculos de mensagens acumuladas para obter as fórmulas exatas.