ApsaraMQ for RabbitMQ:Best practices for instance throttling

Instance total TPS

Choisissez l'approche adaptée à votre profil de trafic :

Single-node SendMessage TPS

ApsaraMQ for RabbitMQ répartit le trafic sur plusieurs nœuds backend. Si le trafic se concentre sur un seul nœud, la limite par nœud peut être atteinte même si la limite au niveau de l'instance n'est pas saturée.

Pour distribuer la charge uniformément entre les nœuds :

• Ouvrez au moins 10 connexions par file d'attente. Chaque connexion peut être routée vers un nœud différent, ce qui permet de répartir la charge d'envoi.

• Pour les utilisateurs de Spring, configurez CachingConnectionFactory en mode CONNECTION. Cela crée une nouvelle connexion par session au lieu de multiplexer les canaux sur une seule connexion. Spring integration.

CloudMonitor (recommandé)

• Granularité : niveau minute (période d'une minute).

• Portée : pic de TPS au niveau de l'instance.

• Coût : gratuit.

• Affiche les tendances de TPS sur les 14 derniers jours.

• Prend en charge les règles d'alerte pour notifier avant que la limitation de débit ne se produise.

Interroger le pic de TPS d'une instance et configurer une règle d'alerte avec CloudMonitor.

Page Instance Details (recommandé)

• Granularité : pic de TPS à la seconde.

• Portée : pic de TPS au niveau de l'instance et par opération API.

• Coût : gratuit.

• Identifie les pics de trafic courts avec une précision à la seconde.

• Prend en charge le filtrage par opération API.

Seuls les résultats des 10 premières minutes sont affichés.

Interroger le pic de TPS d'une instance sur la page Instance Details.

Simple Log Service

• Granularité : pic de TPS à la seconde.

• Portée : pic de TPS au niveau de l'instance.

• Coût : facturé par Simple Log Service (Éléments facturables du paiement à la fonctionnalité).

• Prend en charge les instructions de requête SLS pour une analyse avancée.

Nécessite une familiarité avec la syntaxe des requêtes SLS. Les résultats peuvent être plus difficiles à interpréter que ceux de CloudMonitor ou de la page Instance Details.

Interroger le pic de TPS d'une instance avec Simple Log Service.

Instance total TPS

Single-node SendMessage TPS

Le broker limite le TPS de SendMessage sur chaque nœud backend de l'instance.

Limites par opération API

Ces limites s'appliquent par instance. Les instances de cluster exclusif serverless ne subissent aucune limitation de débit par API.

Code d'erreur et message

Lorsque la limitation de débit est déclenchée, le broker ferme le canal concerné et renvoie :

• Code d'erreur : reply-code=530

• Message d'erreur : reply-text=denied for too many requests

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>
(reply-code=530, reply-text=denied for too many requests, ReqId:5FB4C999314635F952FCBFF6, ErrorHelp[dstQueue=XXX_test_queue,
srcExchange=Producer.ExchangeName,bindingKey=XXX_test_bk, http://mrw.so/6rNqO8], class-id=50, method-id=20)
    at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:516)
    at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:346)
    at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:182)
    at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:114)
    at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:672)
    at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:48)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:599)
    at java.lang.Thread.run(Thread.java:748)

L'erreur inclut l'ID de requête (ReqId), la file d'attente de destination (dstQueue), l'échange source (srcExchange) et la clé de liaison (bindingKey).

Récupération après fermeture de canal

Comme seul le canal est fermé, interceptez AlreadyClosedException et recréez-le. L'exemple Java suivant illustre la récupération basée sur des tentatives :

private static final int MAX_RETRIES = 5; // Maximum retry attempts
private static final long WAIT_TIME_MS = 2000; // Wait time between retries in milliseconds

private void doAnythingWithReopenChannels(Connection connection, Channel channel) {
    try {
        // ......
        // Any operation to be performed in the current channel.
        // For example, sending or consuming messages.
        // ......

    } catch (AlreadyClosedException e) {
        String message = e.getMessage();
        if (isChannelClosed(message)) {
            // Channel was closed by the broker. Recreate it.
            channel = createChannelWithRetry(connection);
            // Continue with other operations after recovery.
            // ......
        } else {
            throw e;
        }
    }
}

private Channel createChannelWithRetry(Connection connection) {
    for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
        try {
            return connection.createChannel();
        } catch (Exception e) {
            System.err.println("Failed to create channel. Attempt " + attempt + " of " + MAX_RETRIES);
            // If channel creation fails (possibly still throttled), wait and retry.
            if (attempt < MAX_RETRIES) {
                try {
                    Thread.sleep(WAIT_TIME_MS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // Restore the interrupted state.
                }
            } else {
                throw new RuntimeException("Exceeded maximum retries to create channel", e);
            }
        }
    }
    throw new RuntimeException("This line should never be reached");
}

private boolean isChannelClosed(String errorMsg) {
    // Check whether the error message contains "channel.close".
    // This covers both error code 530 (throttling) and 541 (internal error).
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        System.out.println("[ChannelClosed] Error details: " + errorMsg);
        return true;
    }
    return false;
}

Points clés :

• La tentative utilise une attente fixe de 2 000 ms. Ajustez cette valeur en fonction de votre profil de trafic.

• La méthode isChannelClosed vérifie la présence de channel.close dans le message d'erreur, ce qui couvre à la fois reply-code=530 (limitation de débit) et d'autres scénarios de fermeture de canal.

• Après la récupération, reprenez les opérations sur le nouvel objet de canal.