Dynamic MQ Consumer Throttling with Alibaba Cloud ACM

When resources become a bottleneck, service frameworks must apply flow control to consumers. Common strategies include static rate limiting based on request rate, dynamic limits based on resource usage, connection‑level control, and concurrency control. In practice, a combination of these methods yields the best results.

In distributed architectures, calls fall into two categories: synchronous RPC (e.g., RESTful, Dubbo, HSF) and asynchronous MQ (e.g., RocketMQ, Kafka). For synchronous calls, flow control typically limits the number of concurrent calls either globally on the provider side or locally on the consumer side. For asynchronous MQ calls, flow control is applied on the subscriber side, either by limiting concurrent message processing or by introducing a consumption delay.

Consumption‑delay flow control works by adding a configurable pause each time a client processes a message, thereby controlling the overall consumption speed. The theoretical maximum consumption rate is calculated as:

MaxRate = 1 / ConsumInterval * ConcurrentThreadNumber

For example, with 20 consumer threads and a 100 ms delay, the maximum concurrent consumption is roughly 200 messages per second.

This method is simple to implement, requires minimal client‑side dependencies, and does not need dynamic thread‑adjustment interfaces.

To achieve global, dynamic flow control in a distributed system, a configuration center (Alibaba Cloud ACM) can distribute flow‑control parameters to all consumers. The following example uses Alibaba Cloud MQ and ACM with Java.

First, define a global variable for the consumption delay and initialize the ACM configuration service:

// Initialize consumption delay (milliseconds)
static int RCV_INTERVAL_TIME = 10000;
// Initialize ACM configuration service
ConfigService.init("acm.aliyun.com", "<tenantId>", "<accessKey>", "<secretKey>");
// Retrieve configuration
String content = ConfigService.getConfig("app.mq.qos", "DEFAULT_GROUP", 6000);
Properties p = new Properties();
try {
    p.load(new StringReader(content));
    RCV_INTERVAL_TIME = Integer.valueOf(p.getProperty("RCV_INTERVAL_TIME"));
} catch (IOException e) {
    e.printStackTrace();
}

Next, add an ACM listener to update the delay when the configuration changes:

ConfigService.addListener("app.mq.qos", "DEFAULT_GROUP", new ConfigChangeListener() {
    public void receiveConfigInfo(String configInfo) {
        Properties p = new Properties();
        try {
            p.load(new StringReader(configInfo));
            RCV_INTERVAL_TIME = Integer.valueOf(p.getProperty("RCV_INTERVAL_TIME"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
});

Finally, implement the MQ consumer logic that respects the delay:

while (rcvIntervalTimeLeft > 0) {
    if (rcvIntervalTimeLeft > RCV_INTERVAL_TIME) {
        rcvIntervalTimeLeft = RCV_INTERVAL_TIME;
    }
    try {
        if (rcvIntervalTimeLeft >= 100) {
            rcvIntervalTimeLeft -= 100;
            Thread.sleep(100);
        } else {
            Thread.sleep(rcvIntervalTimeLeft);
            rcvIntervalTimeLeft = 0;
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}
System.out.println("Receive: " + message);
// Business logic here
doSomething();
return Action.CommitMessage;

Note: the access to RCV_INTERVAL_TIME is intentionally unsynchronized to provoke discussion about thread‑safety.

Experimental results show that adjusting RCV_INTERVAL_TIME directly influences throughput (tpm). With a 100 ms delay, throughput is about 9,000 tpm; with 5,000 ms, it drops to ~200 tpm; with 1,000 ms, it recovers to ~1,100 tpm, confirming the inverse relationship between delay and consumption rate.