ApsaraMQ for RabbitMQ:Spring integration

Connection and Channel Best Practices

ApsaraMQ for RabbitMQ uses a distributed multi-node architecture on the backend. To avoid single-node performance bottlenecks on the server side, follow these best practices when configuring client connections.

• Use CONNECTION mode: Configure the client to establish multiple TCP connections to the server. In CONNECTION mode, a single client distributes connections across multiple backend service nodes, utilizing the resources of multiple nodes in the cluster. We recommend that you configure the number of connections (CONNECTION) per client at the tens level, adjusting based on the client traffic volume.

• Avoid frequent connection establishment and teardown: Do not frequently create and close connections. Frequent connection cycling reduces the efficiency of message sending and consumption. Maintain persistent connections rather than creating them per operation.

• Limit the number of channels per connection: Set the number of channels that can be created on a single connection to no more than 50. Too many channels on a single connection can cause sudden TCP traffic spikes. Combined with occasional packet reordering, this may trigger the Linux Netfilter Conntrack mechanism to classify packets as INVALID, resulting in the client actively sending RST to reset the connection.

You can configure the connection mode and CONNECTION/CHANNEL count using the following parameters:

• spring.rabbitmq.cache.connection.mode - Set to connection for CONNECTION mode.

• spring.rabbitmq.cache.connection.size - Number of connections. Recommended: tens level.

• spring.rabbitmq.cache.channel.size - Number of channels per connection. Recommended: no more than 50.

Produce messages

In RabbitMQService, obtain RabbitTemplate through dependency injection and call its send method to send a message.

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
public class RabbitMQService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange, String routingKey, String content) {
        // Set the MessageId.
        String msgId = UUID.randomUUID().toString();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setMessageId(msgId);
        // Create a Message.
        Message message = new Message(content.getBytes(StandardCharsets.UTF_8), messageProperties);
        /*
         * Call the send() interface to send the message.
         * exchange: the name of the exchange.
         * routingKey: the routing key.
         * message: the message content.
         * correlationData is used for publisher confirms.
         */
        rabbitTemplate.send(exchange, routingKey, message, null);
    }
}

Consume messages

Use the @RabbitListener annotation to consume messages:

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class MessageListener {

     /**
     * Receive a message.
     * @param message The message.
     * @param channel The channel.
     * @throws IOException
     * Replace queues with the name of the queue that you created.
     */
    @RabbitListener(queues = "myQueue")
    public void receiveFromMyQueue(Message message, Channel channel) throws IOException {
        // Enter the business logic for message consumption.
        ...
        // You must return an Ack within the Ack validity period (consumption timeout). Otherwise, the confirmation is invalid and the message is redelivered.
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

The following are common optional configurations for RabbitListener:

Important notes

Verify whether your client uses the receiveAndConvert method of RabbitTemplate for message consumption.

This method accepts parameters such as the queue name and timeout duration. When a timeout is specified, the internal processing logic has the following issue: before consuming a message, the method first sends a BasicCancel command to cancel the consumer subscription, and then sends an Ack to confirm the message.

However, when ApsaraMQ for RabbitMQ receives a BasicCancel request, it immediately cancels the corresponding consumer and requeues all unacknowledged messages. This means the subsequent Ack sent by the client is effectively invalid — the server does not treat it as a valid acknowledgment but instead redelivers these messages, causing duplicate consumption on the client side.

Additionally, if the queue has a high message throughput, frequent cancel and requeue operations may lead to message accumulation, further impacting consumption stability.