Subscription consistency in ApsaraMQ for RocketMQ - ApsaraMQ for RocketMQ

Dans ApsaraMQ for RocketMQ, la cohérence des abonnements signifie que toutes les instances de consommateur identifiées par le même ID de groupe doivent s'abonner aux mêmes rubriques et aux mêmes tags. Des abonnements incohérents peuvent provoquer des erreurs dans la logique de consommation des messages, ainsi que des consommations en double ou manquantes.

Qu'est-ce que la cohérence des abonnements ?

Dans ApsaraMQ for RocketMQ, un groupe de consommateurs est un ensemble d'instances de consommateur. Pour la plupart des applications distribuées, plusieurs instances de consommateur sont associées à un même groupe de consommateurs.

Dans ApsaraMQ for RocketMQ, un abonnement désigne la relation d'abonnement entre un groupe de consommateurs et une rubrique spécifique, y compris les conditions utilisées pour filtrer les messages dans la rubrique abonnée.

Pour garantir la cohérence des abonnements, toutes les instances de consommateur identifiées par le même ID de groupe doivent présenter les mêmes caractéristiques sur les points suivants :

Les rubriques auxquelles les instances de consommateur s'abonnent doivent être identiques.

Par exemple, si le Consommateur 1 s'abonne à la Rubrique A et à la Rubrique B, le Consommateur 2 du même groupe doit également s'abonner à la Rubrique A et à la Rubrique B. Le Consommateur 2 ne peut pas s'abonner uniquement à la Rubrique A ou à la Rubrique B, ni s'abonner à la Rubrique A et à la Rubrique C.

Les tags auxquels les instances de consommateur s'abonnent dans la rubrique doivent être identiques, y compris leur nombre et leur ordre.

Par exemple, si le Consommateur 1 s'abonne à Tag1||Tag2 dans la Rubrique B, le Consommateur 2 du même groupe doit également s'abonner à Tag1||Tag2 dans la Rubrique B. Le Consommateur 2 ne peut pas s'abonner uniquement à Tag1 ou à Tag2, ni s'abonner à Tag2||Tag1 dans la Rubrique B.
Si les instances de consommateur d'un groupe de consommateurs s'abonnent à plusieurs rubriques, le type de message de ces rubriques doit être identique.

Par exemple, si le Consommateur 1 et le Consommateur 2 s'abonnent à la Rubrique A et à la Rubrique B, les rubriques doivent être du même type de message, tel que le type de message normal ou le type de message ordonné. Pour plus d'informations sur les types de messages pris en charge par ApsaraMQ for RocketMQ, consultez Types de messages.

La figure suivante illustre un exemple d'abonnements cohérents. Des groupes ayant des ID différents s'abonnent à des rubriques différentes. Chaque groupe contient les instances de consommateur C1, C2 et C3. Toutes les instances de consommateur d'un même groupe s'abonnent aux mêmes rubriques et aux mêmes tags. 正确订阅关系

Important

ApsaraMQ for RocketMQ vous permet d'utiliser les SDK client TCP et HTTP pour envoyer et recevoir des messages. Outre la garantie de la cohérence des abonnements des instances de consommateur au sein d'un même groupe, vous devez également assurer la cohérence entre la version du protocole du groupe de consommateurs qui s'abonne aux messages et la version du protocole du SDK. Par exemple, si vous utilisez un SDK client TCP pour envoyer et recevoir des messages, vous devez également utiliser un groupe de consommateurs utilisant le protocole TCP pour s'abonner aux messages. Dans le cas contraire, la consommation des messages échouera.

Comment vérifier la cohérence des abonnements des instances de consommateur d'un groupe ?

Pour vérifier la cohérence des abonnements des instances de consommateur d'un groupe, effectuez les opérations suivantes :

Connectez-vous à la console ApsaraMQ for RocketMQ. Sur la page Instances, localisez l'instance que vous souhaitez gérer et cliquez sur son nom pour accéder à la page Instance Details.
Dans le volet de navigation de gauche, cliquez sur Groups. Sur la page qui s'affiche, cliquez sur le nom du groupe de consommateurs que vous souhaitez gérer.
Dans la section Subscriptions de la page Group Details, vérifiez la cohérence des abonnements du groupe de consommateurs.

Abonnement cohérent 1 : Un tag dans une rubrique

Sur la figure ci-dessous, les instances de consommateur C1, C2 et C3 appartiennent au même groupe. Les instances de consommateur s'abonnent à Tag 1 dans la Rubrique A. Ce scénario respecte les exigences de cohérence des abonnements.

** 正确示例1 Exemple de code pour l'abonnement cohérent 1**

Les abonnements de C1, C2 et C3 sont cohérents. Par conséquent, le code écrit pour que C1, C2 et C3 s'abonnent aux messages doit être identique. Exemple de code :

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_test_1");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("TopicA", "Tag1", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

Abonnement cohérent 2 : Plusieurs tags dans une rubrique

Sur la figure ci-dessous, les instances de consommateur C1, C2 et C3 appartiennent au même groupe. Les instances de consommateur s'abonnent à Tag1||Tag2 dans la Rubrique B. Ainsi, toutes les instances de consommateur s'abonnent aux messages portant le Tag 1 ou le Tag 2 dans la Rubrique B, dans le même ordre. Ce scénario respecte les exigences de cohérence des abonnements.

** 正确示例2 Exemple de code pour l'abonnement cohérent 2**

Les abonnements de C1, C2 et C3 sont cohérents. Par conséquent, le code écrit pour que C1, C2 et C3 s'abonnent aux messages doit être identique. Exemple de code :

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_test_2");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("TopicB", "Tag1||Tag2", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

Abonnement cohérent 3 : Plusieurs tags dans plusieurs rubriques

Sur la figure ci-dessous, les instances de consommateur C1, C2 et C3 appartiennent au même groupe. Les instances de consommateur s'abonnent à la Rubrique A et à la Rubrique B. Aucun tag n'est spécifié pour la Rubrique A, tandis que Tag1||Tag2 est spécifié pour la Rubrique B. Dans ce cas, les instances de consommateur s'abonnent à tous les messages de la Rubrique A et aux messages portant le Tag 1 ou le Tag 2 dans la Rubrique B, dans le même ordre. Ce scénario respecte les exigences de cohérence des abonnements.

** 正确示例3 Exemple de code pour l'abonnement cohérent 3**

Les abonnements de C1, C2 et C3 sont cohérents. Par conséquent, le code écrit pour que C1, C2 et C3 s'abonnent aux messages doit être identique. Exemple de code :

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_test_3");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("TopicA", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });     
    consumer.subscribe("TopicB", "Tag1||Tag2", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

Abonnements incohérents

Vous utilisez ApsaraMQ for RocketMQ pour envoyer et recevoir des messages. Les instances de consommateur ne reçoivent pas les messages comme prévu et les abonnements des instances de consommateur affichés dans la console ApsaraMQ for RocketMQ sont incohérents. Cela peut être dû à l'un des problèmes suivants survenant sur les instances de consommateur :

Abonnement incohérent 1 : Les instances de consommateur appartenant au même groupe s'abonnent à des rubriques différentes

Sur la figure ci-dessous, les instances de consommateur C1, C2 et C3 appartiennent au même groupe. Les instances de consommateur s'abonnent à des rubriques différentes. C1 s'abonne à la Rubrique A. C2 s'abonne à la Rubrique B. C3 s'abonne à la Rubrique C. Ce scénario ne respecte pas les exigences de cohérence des abonnements.

Exemple de code pour l'abonnement incohérent 1

Instance de consommateur 1 :

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_test_1");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("TopicA", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

Instance de consommateur 2 :

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_test_1");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("TopicC", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

Instance de consommateur 3 :

    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_test_1");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("TopicB", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

Abonnement incohérent 2 : Les instances de consommateur appartenant au même groupe s'abonnent à des tags différents dans la même rubrique

Sur la figure ci-dessous, les instances de consommateur C1, C2 et C3 appartiennent au même groupe. Les instances de consommateur s'abonnent à la Rubrique A. Cependant, C1 s'abonne à Tag 1 dans la Rubrique A, tandis que C2 et C3 s'abonnent à Tag 2 dans la Rubrique A. Les instances de consommateur s'abonnent à des tags différents dans la même rubrique. Ce scénario ne respecte pas les exigences de cohérence des abonnements.

错误示例2

Exemple de code pour l'abonnement incohérent 2

Instance de consommateur 1 :

 Properties properties = new Properties();
 properties.put(PropertyKeyConst.GROUP_ID, "GID_test_2");
 Consumer consumer = ONSFactory.createConsumer(properties);
 consumer.subscribe("TopicA", "Tag1", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 });

Instance de consommateur 2 :

 Properties properties = new Properties();
 properties.put(PropertyKeyConst.GROUP_ID, "GID_test_2");
 Consumer consumer = ONSFactory.createConsumer(properties);
 consumer.subscribe("TopicA", "Tag2", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 });

Instance de consommateur 3 :

 Properties properties = new Properties();
 properties.put(PropertyKeyConst.GROUP_ID, "GID_test_2");
 Consumer consumer = ONSFactory.createConsumer(properties);
 consumer.subscribe("TopicA", "Tag2", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 });

Abonnement incohérent 3 : Les instances de consommateur appartenant au même groupe s'abonnent à la même rubrique et aux mêmes tags de la rubrique, mais les tags sont spécifiés dans un ordre différent

Sur la figure ci-dessous, les instances de consommateur C1, C2 et C3 appartiennent au même groupe. Les instances de consommateur s'abonnent à la Rubrique A et à la Rubrique B. Aucun tag n'est spécifié pour la Rubrique A. Tag 1 et Tag 2 sont spécifiés pour la Rubrique B. Cependant, C1 s'abonne à Tag1||Tag2 dans la Rubrique B, tandis que C2 et C3 s'abonnent à Tag2||Tag1 dans la Rubrique B. Ce scénario ne respecte pas les exigences de cohérence des abonnements.

错误示例3

Exemple de code pour l'abonnement incohérent 3

Instance de consommateur 1 :

 Properties properties = new Properties();
 properties.put(PropertyKeyConst.GROUP_ID, "GID_test_3");
 Consumer consumer = ONSFactory.createConsumer(properties);
 consumer.subscribe("TopicA", "*", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 }); 
 consumer.subscribe("TopicB", "Tag1||Tag2", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 });

Instance de consommateur 2 :

 Properties properties = new Properties();
 properties.put(PropertyKeyConst.GROUP_ID, "GID_test_3");
 Consumer consumer = ONSFactory.createConsumer(properties);
 consumer.subscribe("TopicA", "*", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 }); 
 consumer.subscribe("TopicB", "Tag2||Tag1", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 });

Instance de consommateur 3 :

 Properties properties = new Properties();
 properties.put(PropertyKeyConst.GROUP_ID, "GID_test_3");
 Consumer consumer = ONSFactory.createConsumer(properties);
 consumer.subscribe("TopicA", "*", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 }); 
 consumer.subscribe("TopicB", "Tag2||Tag1", new MessageListener() {
 public Action consume(Message message, ConsumeContext context) {
 System.out.println(message.getMsgID());
 return Action.CommitMessage;
 }
 });

FAQ

Un groupe de consommateurs peut-il s'abonner à plusieurs rubriques ?

Oui, un groupe de consommateurs peut s'abonner à plusieurs rubriques. Si un groupe de consommateurs s'abonne à une rubrique spécifique, un abonnement est établi. Assurez-vous que toutes les instances de consommateur du groupe de consommateurs s'abonnent à la rubrique.

Puis-je supprimer un abonnement dans la console ApsaraMQ for RocketMQ ?

Non, vous ne pouvez pas supprimer un abonnement dans la console ApsaraMQ for RocketMQ. Vous devez supprimer un abonnement en configurant le paramètre correspondant dans le code du client.