Utilisation du client SDK pour consommer les données suivies depuis une instance PolarDB-X 1.0 - Data Transmission Service

Prérequis

Vous avez créé une instance de suivi des modifications. L'instance est dans l'état Normal. Pour plus d'informations, consultez la page Création d'une tâche de suivi des modifications pour une instance PolarDB-X 1,0 ou Création d'une tâche de suivi des modifications pour une base de données logique DMS.
Vous avez créé un groupe de consommateurs pour votre instance d'abonnement.
Si vous utilisez un utilisateur RAM pour consommer les données souscrites, cet utilisateur doit disposer de l'autorisation AliyunDTSFullAccess ainsi que des autorisations d'accès aux objets souscrits. Pour plus d'informations, consultez les pages Attribution d'autorisations à un utilisateur RAM pour gérer DTS via une stratégie système et Gestion des autorisations d'un utilisateur RAM.

Remarques sur l'utilisation

Lors de la consommation des données souscrites, vous devez impérativement terminer la consommation des données avant d'appeler la méthode commit de l'objet DefaultUserRecord pour valider les informations de décalage (offset). N'appelez pas la méthode de validation avant d'avoir consommé les données, sous peine de provoquer une perte de données.
Les différents processus de consommation sont indépendants les uns des autres.

Procédure

Téléchargez et décompressez le code exemple du SDK.
Vérifiez la version du code du SDK.
1. Accédez au répertoire dans lequel vous avez décompressé le code exemple du SDK.
2. Ouvrez le fichier pom.xml situé dans ce répertoire à l'aide d'un éditeur de texte.
3. Mettez à jour le SDK de suivi des modifications vers la dernière version.
  
  Remarque
  Vous trouverez la dernière dépendance Maven sur la page dts-new-subscribe-sdk.
  Emplacement du paramètre de version du SDK (cliquez pour développer)
  <name>dts-new-subscribe-sdk</name> <url>https://www.aliyun.com/product/dts</url> <description>The Aliyun new Subscribe SDK for Java used for accessing Data Transmission Service</description> <packaging>jar</packaging> <groupId>com.aliyun.dts</groupId> <artifactId>dts-new-subscribe-sdk</artifactId> <version>2.1.4</version>

Modifiez le code du SDK.

Ouvrez le fichier décompressé avec un éditeur de code.
Selon le mode d'utilisation du client SDK, ouvrez le fichier DistributedDTSConsumerDemo.java.

Remarque
Le chemin d'accès au fichier Java est le suivant : aliyun-dts-subscribe-sdk-java-master/src/test/java/com/aliyun/dts/subscribe/clients/.

Définissez les paramètres dans le code Java.

public static void main(String[] args) throws ClientException {
        // Configuration for subscribing to a distributed data source, such as PolarDB-X 1.0 (formerly DRDS). Configure information such as the AccessKey, instance ID, main task ID, and consumer group.
        String accessKeyId = "LTA***********99reZ";
        String accessKeySecret = "****************";
        String regionId = "cn-hangzhou";
        String dtsInstanceId = "dtse5212sed162****";
        String jobId = "l791216x16d****";
        String sid = "dtsip412t13160****";
        String userName = "xftest";
        String password = "******";
        String proxyUrl = "dts-cn-****.com:18001";
        // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
        String checkpoint = "1639620090";

        // Convert physical database/table name to logical database/table name
        boolean mapping = true;
        // if force use config checkpoint when start. for checkpoint reset, only assign mode works
        boolean isForceUseInitCheckpoint = false;

        ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.ASSIGN;
        DistributedDTSConsumerDemo demo = new DistributedDTSConsumerDemo(userName, password, regionId,
                jobId, sid, dtsInstanceId, accessKeyId, accessKeySecret, subscribeMode, proxyUrl,
                checkpoint, isForceUseInitCheckpoint, mapping);
        demo.start();
    }

Paramètre	Description	Comment l'obtenir
accessKeyId	ID de la clé AccessKey.	Pour plus d'informations, consultez la page Obtention d'une paire de clés AccessKey.
accessKeySecret	Clé secrète AccessKey.
regionId	ID de la région où se trouve la tâche de suivi des modifications.	Dans la console DTS, cliquez sur l'ID de l'instance de suivi des modifications cible. Sur la page Basic Information, vous pouvez récupérer les informations relatives à la région. Par exemple, si la région est Chine (Hangzhou), définissez ce paramètre sur `cn-hangzhou`. Pour plus d'informations, consultez la page Liste des régions.
dtsInstanceId	ID de l'instance de suivi des modifications.	Dans la console DTS, cliquez sur l'ID de l'instance de suivi des modifications cible. Sur la page Basic Information, vous pouvez récupérer l'DTS Instance ID de l'instance de suivi des modifications.
jobId	ID de la tâche de suivi des modifications.	Appelez l'opération DescribeDtsJobs pour récupérer l'ID de la tâche de suivi des modifications (DtsJobId).
sid	ID du groupe de consommateurs.	Dans la console DTS, cliquez sur l'ID de l'instance de suivi des modifications cible. Dans le volet de navigation de gauche, cliquez sur Consume Data. Vous pouvez récupérer l'Consumer Group ID/Name et le Account du groupe de consommateurs. Remarque Le mot de passe du compte du groupe de consommateurs est spécifié lors de la création du groupe.
userName	Compte du groupe de consommateurs.
password	Mot de passe du compte du groupe de consommateurs.
proxyUrl	Endpoint et port du canal de suivi des modifications. Remarque Si l'instance ECS sur laquelle vous déployez le client SDK et le canal de suivi des modifications se trouvent dans le même réseau classique ou le même VPC, abonnez-vous aux données via le réseau interne afin d'obtenir la latence la plus faible possible. Nous déconseillons l'utilisation d'un endpoint public en raison d'une éventuelle instabilité du réseau.	Dans la console DTS, cliquez sur l'ID de l'instance de suivi des modifications cible. Sur la page Basic Information, vous pouvez récupérer les informations relatives au Network.
checkpoint	Décalage du consommateur (consumer offset). Il s'agit de l'horodatage à partir duquel le client SDK commence à consommer les enregistrements de données. La valeur correspond à un horodatage UNIX exprimé en secondes. Remarque Vous pouvez utiliser les informations de décalage du consommateur pour les opérations suivantes : En cas d'interruption du processus de consommation, vous pouvez transmettre le décalage du consommateur pour reprendre la consommation des données et éviter toute perte. Lors du démarrage du client SDK, vous pouvez transmettre le décalage du consommateur requis pour ajuster le décalage d'abonnement et consommer les données selon vos besoins.	Le décalage du consommateur doit se situer dans la plage d'horodatage de l'instance de suivi des modifications et doit être converti en horodatage UNIX. Remarque Vous pouvez afficher la plage d'horodatage de l'instance de suivi des modifications dans la colonne Data Range de la liste des tâches de suivi. Vous pouvez utiliser un moteur de recherche pour trouver un convertisseur d'horodatage UNIX.

Facultatif : Pour modifier le type de données des données souscrites, modifiez la méthode buildRecordListener() ou utilisez une classe personnalisée.

public static Map<String, RecordListener> buildRecordListener() {
        // user can impl their own listener
        RecordListener mysqlRecordPrintListener = new RecordListener() {
            @Override
            public void consume(DefaultUserRecord record) {

                OperationType operationType = record.getOperationType();

                if (operationType.equals(OperationType.INSERT)
                        || operationType.equals(OperationType.UPDATE)
                        || operationType.equals(OperationType.DELETE)
                        || operationType.equals(OperationType.HEARTBEAT)) {

                    // consume record
                    RecordListener recordPrintListener = new DefaultRecordPrintListener(DbType.MySQL);

                    recordPrintListener.consume(record);

                    //commit method push the checkpoint update
                    record.commit("");
                }
            }
        };
        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
    }

Ouvrez la structure du projet dans votre IDE et assurez-vous que la version OpenJDK utilisée pour le projet est la 1,8.

Exécutez le code client.

La sortie indique que le client s'abonne aux modifications de données de la base de données source.

Le client SDK collecte et affiche périodiquement des statistiques sur la consommation de données. Ces statistiques incluent le nombre total d'enregistrements de données envoyés et reçus, le volume total de données et le nombre d'enregistrements par seconde (RPS).

Tableau 1. Statistiques de consommation de données

Paramètre

Description

outCounts

Nombre total d'enregistrements de données consommés par le client SDK.

outBytes

Volume total de données consommées par le client SDK, en octets.

outRps

Nombre de requêtes par seconde envoyées par le client SDK pour consommer les données.

outBps

Nombre de bits transmis par seconde lors de la consommation des données par le client SDK.

count

Aucun.

inBytes

Volume total de données envoyées par le serveur DTS, en octets.

DStoreRecordQueue

Taille de la file d'attente de cache de données lorsque le serveur DTS envoie des données.

inCounts

Nombre total d'enregistrements de données envoyés par le serveur DTS.

inRps

Nombre de requêtes envoyées par le serveur DTS par seconde.

inBps

Nombre de bits transmis par seconde lors de l'envoi des données par le serveur DTS.

__dt

Horodatage auquel le client SDK reçoit les données, en millisecondes.

DefaultUserRecordQueue

Taille de la file d'attente de cache de données après sérialisation.