ApsaraMQ for RabbitMQ:SDK FAQ

Can open source RabbitMQ clients connect to ApsaraMQ for RabbitMQ?

Yes. ApsaraMQ for RabbitMQ is fully compatible with open source RabbitMQ, so any open source RabbitMQ client can connect directly.

To authenticate, create a username and password and grant permissions based on the identity verification and permission pattern of your instance. For details, see Permission management.

Which programming languages are supported?

ApsaraMQ for RabbitMQ supports all programming languages and frameworks that open source RabbitMQ SDKs support over AMQP (Advanced Message Queuing Protocol). For the full list, see Supported SDKs.

Can I use the open source RabbitMQ JMS Client?

No. ApsaraMQ for RabbitMQ does not currently support the RabbitMQ JMS (Java Message Service) Client.

How do I reject and requeue messages?

Both basicReject and basicNack require manual acknowledgement mode (autoAck = false). With automatic ACK enabled, the broker considers each message delivered on send, so reject and nack calls have no effect.

With manual ACK, use one of these methods:

• basicReject -- Rejects a single message.

  Parameter	Description
  deliveryTag	Unique identifier for the message delivery within a channel.
  requeue	Set to true to requeue the message, or false to discard it or route it to a dead-letter exchange.

• basicNack -- Rejects one or more messages in a single call.

  Parameter	Description
  deliveryTag	Unique identifier for the message delivery within a channel.
  multiple	Set to true to reject the specified message and all earlier unacknowledged messages. Set to false to reject only the specified message.
  requeue	Set to true to requeue, or false to discard or route to a dead-letter exchange.

  Example: reject and requeue with basicNack

  // Consume from queue "test" with manual acknowledgement (autoAck = false)
  channel.basicConsume("test", false, new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope,
                        AMQP.BasicProperties properties, byte[] body)
              throws IOException {
          System.out.println("Rejected: " + new String(body, "UTF-8")
              + ", deliveryTag: " + envelope.getDeliveryTag()
              + ", messageId: " + properties.getMessageId());
          // Reject this message and all earlier unacknowledged messages, then requeue them
          channel.basicNack(envelope.getDeliveryTag(), true, true);
      }
  });

How do I set a message ID?

Set the message-id property in Basic.Properties on the producer client to assign a unique identifier to each message.

For more about message IDs, see Message ID.

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .messageId("messageid")
    .build();
channel.basicPublish("<ExchangeName>", "RoutingKey", true, props,
    ("Message Body").getBytes(StandardCharsets.UTF_8));

properties = pika.BasicProperties(
    app_id='example-publisher',
    content_type='application/json',
    message_id='messageid'
)

$msg = new AMQPMessage($msgBody, [
    'application_headers' => $amqpTable,
    'content_type' => 'text/plain',
    'delivery_mode' => 2,
    'message_id' => 'messageid',
]);

err = ch.Publish(
    "helloExchange",
    "hello",
    false,
    false,
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
        MessageId:   "messageId",
    },
)

channel.BasicPublish(
    exchange: "",
    routingKey: "hello",
    basicProperties: null,
    body: body
);

How do I delete all messages from a queue?

Call queuePurge in the Java client library to remove every message from a specific queue:

channel.queuePurge("queue-name");

How do I enable TLS-encrypted connections?

By default, clients connect over port 5672 (unencrypted). To encrypt the connection, connect to port 5671 and configure TLS on com.rabbitmq.client.ConnectionFactory:

private void setSSL(com.rabbitmq.client.ConnectionFactory factory)
        throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
    // Use TLS 1.2 for encrypted communication
    SSLContext sslContext = SSLContext.getInstance("TLSv1.2");

    // Initialize the trust manager with the default certificate store
    TrustManagerFactory trustManagerFactory =
        TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
    trustManagerFactory.init((KeyStore) null);

    sslContext.init(null, trustManagerFactory.getTrustManagers(), null);

    factory.useSslProtocol(sslContext);
}

Why do consumers show uneven CPU or memory usage?

This typically happens when client connections are unevenly distributed across backend service nodes.

ApsaraMQ for RabbitMQ uses a distributed deployment with multiple backend nodes. The gateway uses a polling mechanism to distribute connections, and each backend node pulls a similar volume of data due to the storage-compute separation architecture. If one node handles fewer connections than others, each client on that node processes a disproportionately large share of data.

Example: Five clients open five connections to three backend nodes (A, B, C). If the connections land as A:2, B:2, C:1, the single client on node C handles all the data that node C pulls -- roughly the same volume as nodes A and B, but shared across fewer clients.

Solution: Open multiple connections per client (for example, 50 per client node) to distribute connections more evenly across backend nodes. In Spring Boot, configure connection mode for consumption.