Data Transmission Service:Consumir dados de assinatura usando código de exemplo do SDK

Procedimento

O procedimento a seguir demonstra como executar o código de exemplo do SDK para consumir dados de Data Subscription no IntelliJ IDEA (Community Edition 2020.1 para Windows).

1. Crie uma instância de Data Subscription. Para mais detalhes, consulte Criar um canal de Data Subscription para uma instância ApsaraDB RDS for MySQL, Criar um canal de Data Subscription para um cluster PolarDB for MySQL ou Criar um canal de Data Subscription para um banco de dados Oracle.

2. Crie um ou mais grupos de consumidores. Para mais informações, consulte Criar grupos de consumidores.

   Importante

   Ao consumir dados de Data Subscription, chame o método commit de DefaultUserRecord para confirmar os checkpoints. A ausência dessa confirmação pode resultar em consumo duplicado de dados.

3. Use o código de exemplo do SDK conforme as necessidades do seu negócio.

   • Usar o novo pacote SDK de Data Subscription (recomendado)

     1. Abra o IntelliJ IDEA e clique em Create New Project para criar um projeto para sua aplicação.

     2. No projeto, localize o arquivo de modelo de objeto do projeto (POM): pom.xml.

     3. Adicione a seguinte dependência ao arquivo pom.xml:

        <dependency>
            <groupId>com.aliyun.dts</groupId>
            <artifactId>dts-new-subscribe-sdk</artifactId>
            <version>{dts_new_sdk_version}</version>
        </dependency>

        Nota

        Encontre a dependência Maven mais recente na página dts-new-subscribe-sdk.

     4. Para mais informações sobre como usar o novo SDK de assinatura, consulte Usar o código de exemplo.

   • Usar uma versão personalizada do novo SDK de Data Subscription

     1. Baixe o pacote de código de exemplo do SDK e descompacte-o.

        Nota

        Clique em e selecione Download ZIP para baixar o pacote.

     2. Acesse o diretório do código de exemplo do SDK descompactado. Abra o arquivo pom.xml com um editor de texto e atualize o SDK de Data Subscription para a versão mais recente.

        Importante

        Obtenha a versão mais recente do SDK de Data Subscription no site do Maven. Para mais detalhes, consulte a página do Maven para o SDK de Data Subscription.

     3. Abra o IntelliJ IDEA e clique em Open or Import.

     4. Na caixa de diálogo exibida, acesse o diretório do código de exemplo do SDK descompactado, expanda as pastas e localize o arquivo pom.xml.

     5. Na caixa de diálogo exibida, selecione Open as Project.

     6. No IntelliJ IDEA, expanda as pastas. De acordo com o modo de uso do cliente SDK, selecione e clique duas vezes no arquivo Java correspondente: DTSConsumerAssignDemo.java ou DTSConsumerSubscribeDemo.java.

        Nota

        O DTS oferece suporte aos seguintes modos de uso do cliente SDK:

        • Modo ASSIGN: Para garantir a ordem global das mensagens, o DTS atribui apenas uma partição (partição 0) a cada tópico de assinatura. Ao usar um cliente SDK no modo ASSIGN, inicie apenas um cliente.

        • Modo SUBSCRIBE: Para assegurar a ordem global das mensagens, o DTS designa somente uma partição (partição 0) por tópico de assinatura. Se você usar o cliente SDK no modo SUBSCRIBE, poderá iniciar vários clientes SDK em um grupo de consumidores para recuperação de desastres. Caso o cliente ativo falhe, outro cliente SDK será automaticamente atribuído à partição 0 para retomar o consumo.

4. Defina os parâmetros necessários no arquivo Java.

   

   Tabela 1. Parâmetros obrigatórios

   Parâmetro	Descrição	Origem
   brokerUrl	O endpoint e o número da porta da instância de Data Subscription. Nota Se a instância ECS que executa o cliente SDK e a instância de Data Subscription estiverem na mesma rede clássica ou Virtual Private Cloud (VPC), use o endpoint interno para a assinatura visando minimizar a latência de rede. Não recomendamos o uso de endpoint público devido à possível instabilidade da rede.	No console do DTS, clique em ID da instância de Data Subscription desejada. Na página Basic Information, obtenha o endpoint e o número da porta na seção Network.
   topic	O tópico de assinatura da instância.	No console do DTS, clique em ID da instância de Data Subscription alvo. Na página Basic Information, obtenha o Topic na seção Basic Information.
   sid	O ID do grupo de consumidores.	No console do DTS, clique em ID da instância de Data Subscription correspondente e, em seguida, clique em Consume Data. Obtenha o Consumer Group ID e a Account do grupo de consumidores. Nota A senha do nome de usuário do grupo de consumidores é definida durante a criação do grupo.
   userName	O nome de usuário do grupo de consumidores. Aviso Caso não use o cliente fornecido neste tópico, defina o nome de usuário no formato <Username>-<Consumer Group ID>. Exemplo: dtstest-dtsaebpv. Caso contrário, a conexão falhará.
   password	A senha do nome de usuário.
   initCheckpoint	O checkpoint de consumo, especificado como um timestamp UNIX, a partir do qual o cliente SDK começa a consumir dados. Exemplo: 1620962769. Nota As informações de checkpoint de consumo são úteis nos seguintes cenários: Para retomar o consumo e evitar perda de dados após uma interrupção da aplicação, informe o último checkpoint de consumo conhecido. Ao iniciar o cliente, passe um checkpoint de consumo específico para consumir dados a partir de uma posição desejada.	O checkpoint de consumo deve estar dentro do intervalo de dados da instância de Data Subscription (conforme ilustrado na figura) e convertido para um timestamp UNIX. Nota Use um mecanismo de busca para encontrar um conversor de timestamp UNIX.
   ConsumerContext.ConsumerSubscribeMode subscribeMode	O modo de uso do cliente SDK. Valores válidos: ConsumerContext.ConsumerSubscribeMode.ASSIGN: Modo ASSIGN. Apenas um cliente SDK em um grupo de consumidores pode consumir dados de Data Subscription. ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE: Modo SUBSCRIBE. Permite iniciar múltiplos clientes SDK no mesmo grupo de consumidores para fins de recuperação de desastres.	N/A

5. Na barra de navegação superior do IntelliJ IDEA, escolha Run > Run para executar o cliente.

   Nota

   Na primeira execução do cliente, o carregamento e a instalação automática das dependências necessárias podem levar algum tempo.

   • A figura a seguir exibe o resultado. O cliente consome com êxito os dados de alteração do banco de dados de origem.

   • Periodicamente, o cliente SDK agrega e exibe estatísticas sobre o consumo de dados, incluindo a contagem total e o volume de registros enviados e recebidos, além de solicitações por segundo (RPS).

     Tabela 2. Estatísticas de consumo de dados

     Parâmetro	Descrição
     outCounts	Quantidade total de registros de dados consumidos pelo cliente SDK.
     outBytes	Volume total de dados consumidos pelo cliente SDK. Unidade: bytes.
     outRps	Número de solicitações por segundo (RPS) com que o cliente SDK consome dados.
     outBps	Taxa de consumo de dados do cliente SDK, em bits por segundo (bps).
     inBytes	Volume total de dados enviados pelo servidor DTS. Unidade: bytes.
     DStoreRecordQueue	Tamanho da fila interna de cache de dados para registros recebidos do servidor DTS.
     inCounts	Quantidade total de registros de dados enviados pelo servidor DTS.
     inRps	RPS com que o servidor DTS envia dados.
     __dt	Timestamp em que o cliente SDK recebe os dados. Unidade: milissegundos.
     DefaultUserRecordQueue	Tamanho da fila de dados que contém registros prontos para processamento pela aplicação consumidora.

Salvar e consultar checkpoints de consumo

Para iniciar ou retomar o consumo de dados (por exemplo, na primeira execução, reinicialização ou nova tentativa interna), o cliente SDK requer um checkpoint de consumo. A tabela abaixo descreve como gerenciar e consultar checkpoints em diferentes cenários para evitar perda de dados, minimizar o consumo duplicado e permitir o consumo sob demanda.

Persistir checkpoints de consumo

Durante um evento de recuperação de desastres no módulo de coleta de dados incrementais (especialmente no modo SUBSCRIBE), o novo módulo não retém o checkpoint de consumo mais recente do cliente. O cliente pode retomar a partir de um checkpoint mais antigo, resultando em consumo duplicado de dados históricos. Por exemplo, antes da troca, o intervalo de checkpoints do módulo antigo vai de 08:00:00 em 11 de novembro de 2023 até 08:00:00 em 12 de novembro de 2023, e o checkpoint do cliente é 08:00:00 em 12 de novembro de 2023. Após a troca, o intervalo de checkpoints do novo módulo vai de 10:00:00 em 08 de novembro de 2023 até 08:01:00 em 12 de novembro de 2023. O cliente inicia a partir do checkpoint inicial do novo módulo (10:00:00 em 08 de novembro de 2023), causando consumo duplicado.

Para evitar consumo duplicado nesse cenário, configure um armazenamento persistente de checkpoints no cliente. O exemplo a seguir apresenta uma implementação possível que você pode adaptar às suas necessidades.

1. Crie uma classe UserMetaStore que estenda AbstractUserMetaStore.

   Por exemplo, para armazenar informações de checkpoint em um banco de dados MySQL, use o seguinte código Java:

   public class UserMetaStore extends AbstractUserMetaStore {
   
       @Override
       protected void saveData(String groupID, String toStoreJson) {
           Connection con = getConnection();
   			  String sql = "insert into dts_checkpoint(group_id, checkpoint) values(?, ?)";
   
           PreparedStatement pres = null;
           ResultSet rs = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
               pres.setString(2, toStoreJson);
               pres.execute();
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
       }
   
       @Override
       protected String getData(String groupID) {
           Connection con = getConnection();
   			  String sql = "select checkpoint from dts_checkpoint where group_id = ?";
   
           PreparedStatement pres = null;
           ResultSet rs = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
   						ResultSet rs = pres.executeQuery()
                             
               String checkpoint = rs.getString("checkpoint");
             
               return checkpoint;
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
       }
   }
   

2. No arquivo consumerContext.java, configure o meio de armazenamento externo utilizando o método setUserRegisteredStore(new UserMetaStore()).

Perguntas frequentes

• Como resolver problemas de conexão com uma instância de Data Subscription?

  Solucione o problema com base na mensagem de erro. Para mais detalhes, consulte Solução de problemas.

• Em qual formato os checkpoints de consumo são persistidos?

  Os dados de checkpoint de consumo persistidos são armazenados no formato JSON. O checkpoint persistido é um timestamp UNIX que pode ser passado diretamente ao SDK. No exemplo de resposta abaixo, o valor 1700709977 para a chave "timestamp" representa o checkpoint de consumo persistido.

  {"groupID":"dtsglg11d48230***","streamCheckpoint":[{"partition":0,"offset":577989,"topic":"ap_southeast_1_vpc_rm_t4n22s21iysr6****_root_version2","timestamp":1700709977,"info":""}]}

Solução de problemas

ERROR
CheckResult{isOk=false, errMsg='telnet dts-cn-hangzhou.aliyuncs.com:18009
failed, please check the network and if the brokerUrl is correct'}
(com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

telnet real node *** failed, please check the network

ERROR CheckResult{isOk=false, errMsg='build kafka consumer failed, error: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata, probably the user name or password is wrong'} (com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

com.aliyun.dts.subscribe.clients.exception.TimestampSeekException: RecordGenerator:seek timestamp for topic [cn_hangzhou_rm_bp11tv2923n87081s_rdsdt_dtsacct-0] with timestamp [1610249501] failed