Todos os produtos
Search

Este produto

    Function Compute:Use o Function Compute para limpar dados de mensagens no RocketMQ

    Central de documentação

    Function Compute:Use o Function Compute para limpar dados de mensagens no RocketMQ

    Última atualização: Jun 29, 2026

    Use os modelos de limpeza de dados do Function Compute para processar dados de mensagens ou modifique o código do modelo conforme seus requisitos de negócio para atender a necessidades personalizadas. Este tópico usa a divisão de conteúdo de dados no ApsaraMQ for RocketMQ como exemplo para apresentar os tipos de modelos de processamento de mensagens e seu uso.

    Visão geral

    O recurso de limpeza de dados oferece capacidades básicas de operadores com base no Function Compute. Após criar uma tarefa de limpeza de dados de mensagens no ApsaraMQ for RocketMQ, acesse o Function Compute para modificar o código e as configurações da função.

    Operador

    Descrição

    Filtragem de mensagens

    Corresponde ao conteúdo da mensagem com base em expressões regulares e envia as mensagens correspondentes para destinos especificados. Para mais informações, consulte Padrões de eventos para fluxos de eventos.

    Conversão de mensagens

    Converte o conteúdo da mensagem com base em condições de correspondência de strings e envia as mensagens convertidas para destinos especificados. Permite, por exemplo, converter maiúsculas e minúsculas. As mensagens transformadas são enviadas aos destinos definidos. Para mais informações, consulte Transformação de eventos em fluxos de eventos.

    Divisão de conteúdo

    Divide o conteúdo da mensagem com base em expressões regulares e envia as mensagens divididas para destinos especificados.

    Roteamento dinâmico

    Corresponde ao conteúdo da mensagem usando expressões regulares. Envia as mensagens correspondentes para destinos especificados e as não correspondentes para o destino padrão.

    Enriquecimento de conteúdo

    Enriquece o conteúdo da mensagem com base em fontes externas. Por exemplo, se o conteúdo original contiver um AccountID, esse ID consulta o banco de dados para obter a região do cliente. O sistema insira a região no corpo da mensagem de source e o envia para os serviços de destino especificados.

    Mapeamento de conteúdo

    Mapeia o conteúdo da mensagem com base em expressões regulares. O sistema pode mascarar campos sensíveis nas mensagens ou reduzir o tamanho da mensagem ao mínimo necessário.

    Exemplos

    Divisão de conteúdo

    Suponha que seja necessário dividir a mensagem original da lista de alunos [Jack, Male, Class 4|Alice, Female, Class 3|John, Male, Class 4] em três mensagens separadas e enviá-las para os respectivos serviços de destino. Para isso, use o operador de divisão de conteúdo. As mensagens divididas ficam da seguinte forma:

    message:
    [Jack, Male, Class 4]
message:
    [Alice, Female, Class 3]
message:
    [John, Male, Class 4]
    image

    Roteamento dinâmico

    A mensagem a seguir contém informações sobre pastas de dente de três marcas:

    message:
[BrandA, toothpaste, $12.98, 100g
 BrandB, toothpaste, $7.99, 80g
 BrandC, toothpaste, $1.99, 100g]

    Essa lista deve ser enviada para tópicos de destino com base em regras dinâmicas personalizadas. Os itens a seguir descrevem as regras:

    • Envie as mensagens que começam com BrandA para os tópicos BrandA-item-topic e BrandA-discount-topic.

    • Encaminhe as mensagens iniciadas com BrandB para os tópicos BrandB-item-topic e BrandB-discount-topic.

    • Direcione as demais mensagens para o tópico Unknown-brand-topic.

    O código de exemplo a seguir mostra o formato JSON das regras:

    {
  "defaultTopic": "Unknown-brand-topic",
  "rules": [
    {
      "regex": "^BrandA",
      "targetTopics": [
        "BrandA-item-topic",
        "BrandA-discount-topic"
      ]
    },
    {
      "regex": "^BrandB",
      "targetTopics": [
        "BrandB-item-topic",
        "BrandB-discount-topic"
      ]
    }
  ]
}
    image

    Enriquecimento de conteúdo

    Neste exemplo, um bloco CIDR é enriquecido. O código a seguir mostra um exemplo dos logs de acesso de um serviço:

    {
  "accountID": "164901546557****",
  "hostIP": "192.168.XX.XX"
}

    O código a seguir mostra como consultar a source do endereço IP e armazenar o relacionamento de mapeamento em um banco de dados MySQL:

    CREATE TABLE `tb_ip` (
    ->      `IP` VARCHAR(256) NOT NULL,
    ->     `Region` VARCHAR(256) NOT NULL,
    ->      `ISP` VARCHAR(256) NOT NULL,
    ->      PRIMARY KEY (`IP`)
    -> );

    O código a seguir apresenta um exemplo de mensagens processadas:

    {
  "accountID": "164901546557****",
  "hostIP": "192.168.XX.XX",
  "region": "beijing"
}
    image

    Mapeamento de conteúdo

    A mensagem a seguir contém informações cadastrais dos funcionários de uma empresa.

    Zhang San, Employee ID 1, 131 1111 1111
Li Si, Employee ID 2, 132 2222 2222
Wang Wu, Employee ID 3, 133 3333 3333

    Os nomes, IDs e números de telefone dos funcionários na mensagem anterior são confidenciais. Portanto, mascare esses dados antes de enviar a mensagem ao serviço de destino. O código a seguir ilustra essa operação:

    Ja*, Employee ID *, ***********
Ma*, Employee ID *, ***********
Dav*, Employee ID *, ***********
    image

    Procedimento

    1. Crie tópicos da instância do ApsaraMQ for RocketMQ

    1. Faça login no console do ApsaraMQ for RocketMQ. No painel de navegação à esquerda, clique em Instances. Na barra de menu superior, selecione uma região e clique em Create Instance.

    2. No painel Create RocketMQ Instance, selecione uma Instance Version, como 4.0 Series. Em Instance Type, escolha Standard Instance. Insira um nome para a instância, como test, e uma descrição. Em seguida, clique em OK.

    3. Na página Instances, clique na instância desejada. No painel de navegação à esquerda da página de detalhes da instância, clique em Topics e depois em Create Topic.

    4. No painel Create Topic, crie dois tópicos chamados source-topic e target-topic. Preencha o campo Description. Em Message Type, selecione Normal Message. Por fim, clique em OK.

      Nota

      Crie pelo menos dois tópicos: um como fonte de evento para enviar mensagens originais e outro como alvo de evento para receber os dados limpos. Os dois tópicos podem pertencer à mesma instância do RocketMQ ou a instâncias diferentes.

    2. Crie um fluxo de eventos

    1. Acesse o console do EventBridge. No painel de navegação à esquerda, clique em Event Streams. Na barra de menu superior, selecione uma região e clique em Create Event Stream.

    2. Na página Create Event Stream, configure a fonte de eventos, a regra de filtragem, o modelo de limpeza de dados e o alvo de eventos nas etapas Source, Filtering, Transformation e Sink. Depois, clique em Save.

      Source e Sink

      • Para Source, selecione a instância test do ApsaraMQ for RocketMQ e, para Topic, selecione source-topic.

      • Em Sink, escolha a instância test do ApsaraMQ for RocketMQ. Para Topic, selecione target-topic.

      Mantenha os valores padrão para os outros itens de configuração.

      Filtering

      Esta etapa é opcional. Mantenha as configurações padrão e clique em Next.

      ③ Transformation

      • Select Alibaba Cloud Service: Function Compute.

      • Selecione Create Function Template: Uma nova função do FC chamada EventStreaming_Transform_Customized_**** será criada junto com o fluxo de eventos.

      • Function Template: Este exemplo utiliza o modelo Content Splitting. Outros modelos disponíveis incluem Content Mapping, Content Enrichment e Dynamic Routing. Escolha um modelo conforme suas necessidades. Esses modelos fornecem lógica básica de processamento de dados para uso direto ou personalização.

      O nome da função é gerado automaticamente com o formato EventStreaming_Transform_Split_ seguido por um sufixo. O código do modelo decodifica eventos usando UTF-8 e analisa o array de mensagens com ast.literal_eval e json.loads. Quando a configuração for válida, um rótulo verde Valid Configuration aparecerá na parte superior da página.

    3. Testar e verifique

    3,1 Enviar mensagens originais no tópico da instância RocketMQ de source

    1. Faça login no console do Message Queue for Apache RocketMQ, localize o tópico source-topic da instância Source configurada ao criar o fluxo de eventos e, na coluna Actions à direita, clique em Quick Experience.

    2. No painel Start Message Production And Consumption, insira a mensagem original [Jack, Male, Class 4|Alice, Female, Class 3|John, Male, Class 4] e clique em OK para enviar a mensagem.

    3,2 Confirme que as mensagens foram divididas corretamente no tópico da instância RocketMQ de destino

    1. No console do Message Queue for Apache RocketMQ, localize o tópico target-topic da instância Sink configurada ao criar o fluxo de eventos, clique no nome do tópico e selecione a aba Message Query.

    2. Em Query Method, selecione Query By Topic e clique em Search. Os resultados da consulta mostram que a mensagem original foi dividida em três mensagens separadas no sink. Clique em Details na linha de cada mensagem. As três mensagens são "data": "Jack, Male, Class 4" , "data": "Alice, Female, Class 3" e "data": "John, Male, Class 4".

    4. Limpar recursos

    Após os testes, caso não precise usar este recurso a curto prazo, libere os recursos criados prontamente para evitar custos desnecessários. Para mais informações, consulte Excluir tópicos, Excluir instâncias do RocketMQ e Excluir funções.