Après avoir configuré un canal de suivi des modifications en créant une tâche de suivi et un groupe de consommateurs, utilisez le kit de développement logiciel (SDK) fourni par DTS pour consommer les données souscrites. Cette rubrique explique comment utiliser l'exemple de code.
Pour consommer les données souscrites à partir d'une source de données PolarDB-X 1,0, consultez Utiliser un SDK pour consommer les données souscrites à partir de PolarDB-X 1,0.
Cette rubrique fournit un exemple de code pour un client SDK en Java. Pour des exemples de code en Python et Go, consultez dts-subscribe-demo.
Prérequis
-
Une instance d'abonnement est créée et s'exécute dans l'état Normal.
RemarquePour obtenir des instructions sur la création d'une instance d'abonnement, consultez Présentation du plan d'abonnement.
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 Accorder des autorisations à un utilisateur RAM pour gérer DTS à l'aide d'une stratégie système et Gérer les autorisations d'un utilisateur RAM.
Précautions
Lorsque vous consommez des données souscrites, terminez la consommation des données avant d'appeler la méthode
commitdeDefaultUserRecordpour valider les informations de décalage. N'appelez pas la méthode commit 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.
Dans la console, Current Offset indique le décalage auquel la tâche de suivi a souscrit, et non le décalage validé par le client.
Procédure
Téléchargez les fichiers d'exemple de code du SDK et décompressez le package.
-
Vérifiez la version du code du SDK.
Accédez au répertoire dans lequel vous avez décompressé l'exemple de code du SDK.
Ouvrez le fichier pom.xml situé dans ce répertoire à l'aide d'un éditeur de texte.
-
Mettez à jour le SDK de suivi des modifications vers la dernière version.
RemarqueVous trouverez la dernière dépendance Maven sur la page dts-new-subscribe-sdk.
-
Modifiez le code du SDK.
Ouvrez les fichiers décompressés à l'aide d'un environnement de développement intégré (IDE).
-
Ouvrez le fichier Java correspondant au mode dans lequel vous souhaitez utiliser le client SDK.
RemarqueLe chemin d'accès au fichier Java est
aliyun-dts-subscribe-sdk-java-master/src/test/java/com/aliyun/dts/subscribe/clients/.Mode d'utilisation
Fichier Java
Description
Scénarios
Mode ASSIGN
DTSConsumerAssignDemo.java
Pour garantir l'ordre global des messages, DTS n'assigne qu'une seule partition (partition 0) à chaque sujet de suivi. Si vous utilisez le client SDK en mode ASSIGN, ne démarrez qu'un seul client SDK.
Un seul client SDK dans un groupe de consommateurs consomme les données souscrites.
Mode SUBSCRIBE
DTSConsumerSubscribeDemo.java
Pour garantir l'ordre global des messages, DTS n'assigne qu'une seule partition (partition 0) à chaque sujet de suivi. Si vous utilisez le client SDK en mode SUBSCRIBE, vous pouvez démarrer plusieurs clients SDK au sein du même groupe de consommateurs pour la reprise après sinistre. Si le client qui consomme les données échoue, un autre client SDK est automatiquement et aléatoirement assigné à la partition 0 pour poursuivre la consommation.
Plusieurs clients SDK au sein du même groupe de consommateurs consomment les données souscrites. Il s'agit d'un scénario de reprise après sinistre des données.
-
Définissez les paramètres dans le code Java.
Paramètre
Description
Comment l'obtenir
brokerUrlSpécifie l'adresse réseau et le numéro de port du canal de suivi des modifications.
RemarqueSi le serveur sur lequel vous déployez le client SDK, tel qu'une instance ECS, et l'instance de suivi des modifications se trouvent dans le même cloud privé virtuel (VPC), vous pouvez consommer les données via le VPC afin de réduire la latence réseau.
Nous déconseillons l'utilisation d'un endpoint public en raison d'une instabilité réseau potentielle.
Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. Sur la page Basic Information, récupérez l'adresse réseau et le numéro de port dans la section Network.
topicLe sujet du canal de suivi des modifications.
Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. Sur la page Basic Information, accédez à la section Basic Information et récupérez le Topic.
sidL'ID du groupe de consommateurs.
Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. Sur la page Consume Data, récupérez l'Consumer Group ID/Name et le Account.
userNameLe nom d'utilisateur du groupe de consommateurs.
AvertissementSi vous n'utilisez pas le client fourni dans cette rubrique, définissez le nom d'utilisateur au format
<nom d'utilisateur du groupe de consommateurs>-<ID du groupe de consommateurs>(par exemple,dtstest-dtsaebpv). Sinon, la connexion échouera.passwordLe mot de passe du compte.
Le mot de passe que vous avez défini pour le nom d'utilisateur du groupe de consommateurs lors de la création du groupe de consommateurs.
initCheckpointLe décalage du consommateur. Il s'agit de l'horodatage UNIX du premier enregistrement de données que le client SDK doit consommer, par exemple 1620962769.
RemarqueVous pouvez utiliser le décalage du consommateur pour :
Reprendre la consommation des données à partir d'un décalage spécifique après une interruption de votre application afin d'éviter toute perte de données.
Ajuster le décalage de départ pour la consommation des données selon vos besoins.
Le décalage du consommateur doit se situer dans la plage d'horodatage de l'instance de suivi et doit être converti en horodatage UNIX.
RemarqueLa colonne Data Range de la liste des tâches de suivi affiche la plage d'horodatage de l'instance d'abonnement cible.
subscribeModeLe mode d'utilisation du client SDK. Vous n'avez pas besoin de modifier ce paramètre.
ConsumerContext.ConsumerSubscribeMode.ASSIGN: mode ASSIGN.ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE: mode SUBSCRIBE.
S/O
Ouvrez la structure du projet dans votre IDE et assurez-vous que la version OpenJDK du projet est la 1,8.
-
Exécutez le code client.
RemarqueLors de la première exécution du code, l'IDE peut mettre un certain temps à charger automatiquement les plugins et les dépendances requis.
Le client SDK collecte et affiche périodiquement des statistiques sur la consommation des 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).
[2025-02-25 18:22:18.160] [INFO ] [subscribe-logMetricsReporter-1-thread-1] [log.metrics:184] - {"outCounts":0.0,"outBytes":0.0,"outRps":0.0,"outBps":0.0,"count":11.0,"inBytes":0.0,"DStoreRecordQueue":0.0,"inCounts":0.0,"inRps":0.0,"inBps":0.0,"__dt":174047893****,"DefaultUserRecordQueue":0.0}Paramètre
Description
outCountsLe nombre total d'enregistrements de données consommés par le client SDK.
outBytesLe volume total de données consommées par le client SDK, en octets.
outRpsLe nombre de requêtes par seconde pour la consommation des données par le client SDK.
outBpsLe nombre de bits transmis par seconde pour la consommation des données par le client SDK.
countLe nombre total de paramètres dans les informations de consommation des données (métriques).
RemarqueCela n'inclut pas
countlui-même.inBytesLe volume total de données envoyées par le serveur DTS, en octets.
DStoreRecordQueueLa taille actuelle de la file d'attente de cache des données lorsque le serveur DTS envoie des données.
inCountsLe nombre total d'enregistrements de données envoyés par le serveur DTS.
inBpsLe nombre de bits transmis par seconde lorsque le serveur DTS envoie des données.
inRpsLe nombre de requêtes par seconde lorsque le serveur DTS envoie des données.
__dtL'horodatage auquel le client SDK reçoit les données, en millisecondes.
DefaultUserRecordQueueLa taille de la file d'attente de cache des données après sérialisation.
-
Modifiez le code pour consommer les données souscrites selon vos besoins.
Lorsque vous consommez des données souscrites, vous devez gérer les décalages des consommateurs afin d'éviter toute perte de données, de minimiser la duplication des données et de permettre une consommation à la demande.
-
Que faire si je ne parviens pas à me connecter à une instance d'abonnement ?
Dépannez le problème en fonction du message d'erreur. Pour plus d'informations, consultez Dépannage.
-
Quel est le format des données d'un décalage de consommateur après persistance ?
Après la persistance d'un décalage de consommateur, les données sont renvoyées au format JSON. Le décalage de consommateur persistant est un horodatage Unix que vous pouvez transmettre directement au SDK. Par exemple, dans les données renvoyées, la valeur
1700709977pour la clé"timestamp"correspond au décalage de consommateur persistant.{"groupID":"dtsglg11d48230***","streamCheckpoint":[{"partition":0,"offset":577989,"topic":"ap_southeast_1_vpc_rm_t4n22s21iysr6****_root_version2","timestamp":170070****,"info":""}]} -
Une tâche de suivi peut-elle être consommée par plusieurs clients en parallèle ?
Non. Bien que le mode SUBSCRIBE permette à plusieurs clients de s'exécuter en parallèle, un seul client peut consommer des données à la fois.
-
Quelle version du client Kafka est encapsulée dans le code du SDK ?
Les versions 2.0.0 et ultérieures de dts-new-subscribe-sdk encapsulent le client Kafka (kafka-clients) 2.7.0. Les versions antérieures à 2.0.0 encapsulent le client Kafka 1.0.0.
RemarqueSi vous utilisez un outil de détection des vulnérabilités des packages de dépendances dans votre processus de développement d'applications et constatez que le client Kafka (kafka-clients) encapsulé par dts-new-subscribe-sdk présente une vulnérabilité de sécurité, vous pouvez résoudre cette vulnérabilité en remplaçant le client par la version
2.1.4-shaded.<dependency> <groupId>com.aliyun.dts</groupId> <artifactId>dts-new-subscribe-sdk</artifactId> <version>2.1.4-shaded</version> </dependency> Étant donné que le client SDK enregistre le décalage des messages toutes les 5 secondes et le valide auprès du serveur DTS, vous pouvez interroger le dernier décalage de consommateur à partir de l'un des emplacements suivants :
Le fichier localCheckpointStore sur le serveur où se trouve le client SDK.
La page Data Consumption du canal de suivi des modifications.
Si vous avez configuré un support de stockage partagé externe persistant, tel qu'une base de données, dans le fichier consumerContext.java à l'aide de
setUserRegisteredStore(new UserMetaStore()), ce support de stockage enregistre le décalage des messages toutes les 5 secondes pour vous permettre de l'interroger.Le support de stockage externe que vous avez configuré à l'aide de
setUserRegisteredStore(new UserMetaStore())dans le fichier consumerContext.java.Le fichier localCheckpointStore sur le serveur où se trouve le client SDK.
Le décalage enregistré sur le serveur DTS (DStore).
L'horodatage de début que vous transmettez à
initCheckpointdans le fichier DTSConsumerSubscribeDemo.java.Le support de stockage externe que vous avez configuré dans le fichier consumerContext.java à l'aide de
setUserRegisteredStore(new UserMetaStore()).Le décalage enregistré sur le serveur DTS (module d'ingestion incrémentielle des données).
RemarqueCe décalage n'est mis à jour qu'après que le client SDK a appelé la méthode commit pour mettre à jour le décalage du consommateur.
L'horodatage de début que vous transmettez à
initCheckpointdans le fichier DTSConsumerSubscribeDemo.java.Le décalage de début du serveur DTS (nouveau module d'ingestion incrémentielle des données).
ImportantSi le module d'ingestion incrémentielle des données est basculé, le nouveau module ne peut pas enregistrer le dernier décalage du consommateur du client. Cela peut entraîner le démarrage de la consommation des données à partir d'un décalage plus ancien. Nous vous recommandons de stocker de manière persistante le décalage du consommateur sur le client.
Lorsqu'elle est définie sur
true, le client SDK utilisera le initCheckpoint transmis comme décalage du consommateur à chaque redémarrage.Lorsqu'elle est configurée sur
falseou non configurée, recherchez le décalage du consommateur de l'enregistrement précédent dans l'ordre suivant :Le support de stockage externe que vous avez configuré dans le fichier consumerContext.java à l'aide de
setUserRegisteredStore(new UserMetaStore()).Le fichier localCheckpointStore sur le serveur où se trouve le client SDK.
Le décalage enregistré sur le serveur DTS (module d'ingestion incrémentielle des données).
RemarqueCe décalage n'est mis à jour qu'après que le client SDK a appelé la méthode commit pour mettre à jour le décalage du consommateur.
Le support de stockage externe configuré dans le fichier consumerContext.java avec
setUserRegisteredStore(new UserMetaStore()).Le décalage enregistré sur le serveur DTS (module d'ingestion incrémentielle des données).
RemarqueCe décalage n'est mis à jour qu'après que le client SDK a appelé la méthode commit pour mettre à jour le décalage du consommateur.
L'horodatage de début que vous transmettez à
initCheckpointdans le fichier DTSConsumerSubscribeDemo.java.Le décalage de début du serveur DTS (nouveau module d'ingestion incrémentielle des données).
-
Créez une méthode
UserMetaStore()qui hérite et implémente la méthodeAbstractUserMetaStore().Par exemple, vous pouvez utiliser une base de données MySQL pour stocker les informations de décalage. L'exemple de code Java suivant montre comment procéder :
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); rs = pres.executeQuery(); if (rs.next()) { String checkpoint = rs.getString("checkpoint"); return checkpoint; } } catch (Exception e) { e.printStackTrace(); } finally { close(rs, pres, con); } return null; } } Dans le fichier consumerContext.java, appelez la méthode
setUserRegisteredStore(new UserMetaStore())pour configurer le support de stockage externe.Analysez la raison du ralentissement de la consommation des données en interrogeant les paramètres des informations statistiques concernant la taille des files d'attente DStoreRecordQueue et DefaultUserRecordQueue.
Si la valeur de DStoreRecordQueue reste à 0, la vitesse à laquelle le serveur DTS extrait les données est lente.
Si la valeur de DefaultUserRecordQueue reste à la valeur par défaut de 512, la vitesse à laquelle le client SDK consomme les données est lente.
Réinitialisez le décalage en modifiant le décalage du consommateur (initCheckpoint) dans le code selon vos besoins.
FAQ
Annexe
Gérer les décalages des consommateurs
Lors du premier démarrage du client SDK, de son redémarrage ou de nouvelles tentatives internes, vous devez interroger et transmettre un décalage de consommateur pour démarrer ou reprendre la consommation des données. Le décalage de consommateur correspond à l'horodatage UNIX du premier enregistrement de données que le client SDK va consommer.
Pour réinitialiser le décalage du consommateur du client, vous pouvez interroger et modifier le décalage en fonction du mode de consommation (mode d'utilisation du SDK) comme décrit dans le tableau suivant.
Les niveaux de stockage des points de contrôle listés ci-dessus (stockage externe, fichier localCheckpointStore et décalage DTS Server/DStore) sont tous écrits lors d'une seule opération de validation, de sorte que leur contenu est presque identique. La principale différence réside dans le fait que le fichier localCheckpointStore est perdu si le nœud ou le conteneur est détruit, tandis que le décalage DTS Server (DStore) est persisté côté serveur.
Scénario | Mode d'utilisation du SDK | Méthode de gestion du décalage |
Interroger un décalage de consommateur | Mode ASSIGN, mode SUBSCRIBE | |
Le client SDK démarre pour la première fois. Vous devez transmettre un décalage de consommateur pour consommer les données. | Mode ASSIGN, mode SUBSCRIBE | Selon le mode d'utilisation du client SDK, sélectionnez le fichier Java DTSConsumerAssignDemo.java ou DTSConsumerSubscribeDemo.java, et configurez le décalage du consommateur ( |
Le client SDK effectue de nouvelles tentatives en interne. Vous devez transmettre le dernier décalage de consommateur enregistré pour reprendre la consommation des données. | Mode ASSIGN | Recherchez le dernier décalage de consommateur enregistré dans l'ordre suivant. Si un décalage est trouvé, les informations de décalage sont renvoyées : |
Mode SUBSCRIBE | Recherchez le dernier décalage de consommateur enregistré dans l'ordre suivant. Si un décalage est trouvé, les informations de décalage sont renvoyées : | |
Le client SDK est redémarré. Vous devez transmettre le dernier décalage de consommateur enregistré pour reprendre la consommation des données. | Mode ASSIGN | En fonction de la configuration |
Mode SUBSCRIBE | Dans ce mode, dans le fichier consumerContext.java, la configuration |
Stocker le décalage du consommateur avec persistance
Si un basculement de reprise après sinistre se produit pour le module d'ingestion incrémentielle des données, le nouveau module ne peut pas enregistrer le dernier décalage du consommateur du client. Cela est particulièrement vrai en mode SUBSCRIBE. Par conséquent, le client peut commencer à consommer les données à partir d'un décalage antérieur, entraînant une consommation répétée des données historiques. Par exemple, supposons qu'avant un basculement de service, la plage de décalage de l'ancien module s'étendait du 11 novembre 2023 à 08:00:00 au 12 novembre 2023 à 08:00:00, et que le décalage du consommateur du client était le 12 novembre 2023 à 08:00:00. Après le basculement, la plage de décalage du nouveau module s'étend du 8 novembre 2023 à 10:00:00 au 12 novembre 2023 à 08:01:00. Dans ce scénario, le client commence la consommation à partir du décalage de début du nouveau module (8 novembre 2023 à 10:00:00), ce qui entraîne une consommation répétée des données.
Pour éviter la consommation répétée des données historiques dans ce scénario de basculement, nous vous recommandons de configurer une méthode de stockage persistant pour le décalage du consommateur sur le client. La méthode d'exemple suivante est fournie à titre de référence. Vous pouvez la modifier selon vos besoins.
Dépannage
Exception | Message d'erreur | Cause | Solution |
Échec de la connexion | | Le | Saisissez le |
| Vous ne pouvez pas vous connecter à l'adresse IP réelle à l'aide de l'adresse du broker. | ||
| Le nom d'utilisateur ou le mot de passe est incorrect. | ||
| Dans le fichier consumerContext.java, | Saisissez un décalage de consommateur situé dans la plage d'horodatage de l'instance d'abonnement. Pour plus d'informations sur la méthode d'interrogation, consultez Description des paramètres. | |
Ralentissement de la consommation de l'abonnement | S/O | ||