Configurez ces paramètres lorsque vous utilisez la version Community du SDK Java pour vous connecter à ApsaraMQ for RocketMQ. Les paramètres sont répartis en trois groupes : connexion, producteur et consommateur.
Paramètres de connexion
Définissez ces paramètres pour chaque client (producteur et consommateur).
| Parameter | Description | Default |
|---|---|---|
NAMESRV_ADDR |
Endpoint TCP de votre instance ApsaraMQ for RocketMQ. Récupérez cette information depuis la page Instance Details dans la console ApsaraMQ for RocketMQ. | Aucune (obligatoire) |
AccessKey |
ID AccessKey pour l'authentification. Consultez la rubrique Créer une paire de clés AccessKey. | Aucune (obligatoire) |
SecretKey |
Secret AccessKey pour l'authentification. Consultez la rubrique Créer une paire de clés AccessKey. | Aucune (obligatoire) |
AccessChannel |
Définissez la valeur sur CLOUD pour activer le traçage des messages cloud. |
Aucune |
Tous les clients nécessitent les paramètres NAMESRV_ADDR, AccessKey et SecretKey. Pour la procédure de configuration complète, consultez la rubrique Préparer l'environnement.
Paramètres du producteur
| Parameter | Description | Default | Valid values |
|---|---|---|---|
producerGroup |
ID du groupe de producteurs. Les producteurs d'une même application qui envoient les mêmes messages appartiennent à un seul groupe. Récupérez l'ID du groupe depuis la console ApsaraMQ for RocketMQ. Consultez la rubrique Termes. | Aucune (obligatoire) | -- |
sendMsgTimeout |
Délai d'attente pour l'envoi d'un message, en millisecondes. | -- | -- |
compressMsgBodyOverHowmuch |
Seuil de taille du corps du message déclenchant la compression automatique, en Ko. Les messages dépassant cette valeur sont compressés avant l'envoi et décompressés côté consommateur. | 4 Ko | -- |
retryTimesWhenSendFailed |
Nombre maximal de tentatives de renvoi en cas d'échec des envois synchrones. Ne s'applique pas aux envois asynchrones. | -- | -- |
maxMessageSize |
Taille maximale du message autorisée par le client, en Mo. Les messages dépassant cette limite déclenchent une erreur. Le broker appliquant également sa propre limite de taille, configurez conjointement les paramètres du client et du broker. | 4 Mo | -- |
Exemple : configurer un producteur
DefaultMQProducer producer = new DefaultMQProducer("<your-producer-group-id>");
producer.setNamesrvAddr("<your-tcp-endpoint>");
// Timeout and retry
producer.setSendMsgTimeout(3000); // 3 seconds
producer.setRetryTimesWhenSendFailed(3); // Retry up to 3 times (synchronous sends only)
// Compression and size limits
producer.setCompressMsgBodyOverHowmuch(4096); // Compress messages over 4 KB
producer.setMaxMessageSize(4 * 1024 * 1024); // 4 MB max message size
producer.start();Le paramètre retryTimesWhenSendFailed s'applique uniquement aux envois synchrones.
Paramètres du consommateur
Offset et pool de threads
| Parameter | Description | Default | Valid values |
|---|---|---|---|
consumerGroup |
ID du groupe de consommateurs. Les consommateurs d'une même application qui s'abonnent aux mêmes messages et utilisent la même logique de consommation appartiennent à un seul groupe. Récupérez l'ID du groupe depuis la console ApsaraMQ for RocketMQ. Consultez la rubrique Termes. | Aucune (obligatoire) | -- |
consumeFromWhere |
Offset de départ lors du lancement d'un nouveau groupe de consommateurs. | Dernier offset | -- |
consumeThreadMin |
Nombre minimal de threads dans le pool de threads du consommateur. | 20 | -- |
consumeThreadMax |
Nombre maximal de threads dans le pool de threads du consommateur. Doit être égal à consumeThreadMin. |
20 | -- |
Contrôle de flux et nouvelles tentatives
| Parameter | Description | Default | Valid values |
|---|---|---|---|
consumeConcurrentlyMaxSpan |
Écart d'offset maximal pour la consommation simultanée dans une seule file d'attente. | 2000 | 1–65535 |
pullThresholdForQueue |
Nombre maximal de messages mis en cache localement par file d'attente. | 1000 | 1–65535 |
pullThresholdSizeForQueue |
Taille totale maximale des messages mis en cache localement par file d'attente, en Mo. | 100 Mo | 1–1024 |
maxReconsumeTimes |
Nombre maximal de tentatives de re-consommation pour un message ayant échoué. | 16 | -- |
suspendCurrentQueueTimeMillis |
Intervalle minimal entre les nouvelles tentatives consécutives de messages ordonnés, en millisecondes. | 1000 ms | 10–30000 |
Exemple : configurer un consommateur push
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("<your-consumer-group-id>");
consumer.setNamesrvAddr("<your-tcp-endpoint>");
// Start consuming from the latest offset
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
// Thread pool: min and max must match
consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(20);
// Flow control
consumer.setConsumeConcurrentlyMaxSpan(2000);
consumer.setPullThresholdForQueue(1000);
consumer.setPullThresholdSizeForQueue(100); // 100 MB per queue
// Retry
consumer.setMaxReconsumeTimes(16);
consumer.subscribe("<your-topic>", "*");
consumer.registerMessageListener((msgs, context) -> {
// Process messages
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();Les paramètresconsumeThreadMinetconsumeThreadMaxdoivent avoir la même valeur.