edit-icon download-icon

MQ client traffic control design

Last Updated: May 28, 2018

MQ (Message Queue) is a commonly-used asynchronous RPC technology. This article takes Alibaba Cloud MQ as an example to explain how to use ACM to implement traffic control over MQ.

Brief introduction to MQ traffic control

For MQ calling, a typical traffic control method is to control traffic at the subscription end. Two traffic control methods are supported:

  • Concurrent traffic control over message subscribers
  • Consumption delay traffic control over message subscribers

The basic principle for consumption delay traffic control over message subscribers is to control the consumption speed by adding a delay upon each consumption at the client end. Under this circumstance, the fastest theoretical concurrent consumption speed is:

  1. MaxRate = 1 / ConsumInterval * ConcurrentThreadNumber

For example, if the concurrent thread number (ConcurrentThreadNumber) is 20, and the consumption delay (ConsumInterval) is 100 ms, then according to the preceding formula:

  1. 200 = 1 / 0.1 * 20

Theoretically, we can limit concurrent consumption traffic within 200.

In comparison with the concurrent thread number traffic control, consumption delay traffic control has some advantages, such as that it’s easier to implement, it’s less dependent on MQ client packages, and it doesn’t need the client to provide the dynamic adjustment API that controls the concurrent thread number.

When using the above traffic control methods, if you want to implement dynamic global control under a distributed architecture, you can simply distribute traffic control parameters from the configuration center.

The following part elaborates how to implement dynamic global traffic control for asynchronous message consumption from the configuration center. The example involves Alibaba Cloud’s MQ (Message Queue) and ACM (Application Configuration Management), and is based on Java.

Note: The reason we take MQ as an example is that MQ Consumer Client SDK currently doesn’t support dynamic adjustment of the existing concurrent thread number, and we can solve the problem of dynamic MQ consumption traffic control by dynamically adjusting consumption delay with ACM.

Basic principles of consumption-delay-based traffic control

As shown in the following diagram, the administrator or application publishes the consumption interval configuration (RCV_INTERVAL_TIME) through the ACM console, which is subscribed to by all MQ consumption applications. Theoretically, it takes no more than 1 second for this configuration to be published and distributed to all clients (depends on network latency).

How It Works

Sample code

This section provides the example code of dynamic global traffic control for asynchronous message consumption from the configuration center. For more information about SDK, see official documentation of MQ and ACM.

Create ACM configuration

Create consumption delay parameters on ACM.

Create Consumption Interval

Set global consumption delay variable

  1. Set global variable for consumption receipt delay.

    1. // Initialize delay parameter for message receipt in milliseconds
    2. static int RCV_INTERVAL_TIME = 10000;
    3. // Initialize configuration service, and then the console will automatically retrieve the following parameters with the example code
    4. ConfigService.init("acm.aliyun.com", /*Tenant ID*/"xxx", /*AK*/"xxx", /*SK*/"yyy");
    5. // Automatically retrieve configuration
    6. String content = ConfigService.getConfig("app.mq.qos", "DEFAULT_GROUP", 6000);
    7. Properties p = new Properties();
    8. try {
    9. p.load(new StringReader(content));
    10. RCV_INTERVAL_TIME = Integer.valueOf(p.getProperty("RCV_INTERVAL_TIME"));
    11. } catch (IOException e) {
    12. e.printStackTrace();
    13. }
  2. Set ACM listener, and ensure the RCV_INTERVAL_TIME parameter is promptly updated upon modification of the configuration.

    1. // Add a listener to the configuration during initialization, which will send a callback notice upon configuration change.
    2. ConfigService.addListener("app.mq.qos", "DEFAULT_GROUP", new ConfigChangeListener() {
    3. public void receiveConfigInfo(String configInfo) {
    4. Properties p = new Properties();
    5. try {
    6. p.load(new StringReader(configInfo));
    7. RCV_INTERVAL_TIME = Integer.valueOf(p.getProperty("RCV_INTERVAL_TIME"));
    8. } catch (IOException e) {
    9. e.printStackTrace();
    10. }
    11. }
    12. });

Set MQ consumption delay logic

The complete example is as follows:


  • In this example, the access of the RCV_INTERVAL_TIME parameter is intentionally kept unlocked, and the reason will not be elaborated here.
  • Aliyun ONS Client doesn’t provide a dynamic concurrent thread number. The default concurrency is 20 by default. Therefore, in this example, we will use the consumption delay parameter to dynamically adjust QoS.
  1. //The following code can be directly pasted into the Main() parameter
  2. Properties properties = new Properties();
  3. properties.put(PropertyKeyConst.ConsumerId, "CID_consumer_group");
  4. properties.put(PropertyKeyConst.AccessKey,"xxx");
  5. properties.put(PropertyKeyConst.SecretKey, "yyy");
  6. properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
  7. // Set TCP access domain name (in this example, we use a public cloud production environment)
  8. properties.put(PropertyKeyConst.ONSAddr,
  9. "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet");
  10. Consumer consumer = ONSFactory.createConsumer(properties);
  11. consumer.subscribe(/*Topic*/"topic-name", /*Tag*/null, new MessageListener()
  12. {
  13. public Action consume(Message message, ConsumeContext context) {
  14. // MQ Subscribe QoS logical start,
  15. // Each consuming process will sleep for RCV_INTERVAL_TIME seconds with 100 ms sleeping cycle.
  16. // Within each cycle, the thread will check RCV_INTERVAL_TIME in case it's set to a smaller value.
  17. // RCV_INTERVAL_TIME <= 0 means no sleeping.
  18. int rcvIntervalTimeLeft = RCV_INTERVAL_TIME;
  19. while (rcvIntervalTimeLeft > 0) {
  20. if (rcvIntervalTimeLeft > RCV_INTERVAL_TIME) {
  21. rcvIntervalTimeLeft = RCV_INTERVAL_TIME;
  22. }
  23. try {
  24. if (rcvIntervalTimeLeft >= 100) {
  25. rcvIntervalTimeLeft -= 100;
  26. Thread.sleep(100);
  27. } else {
  28. Thread.sleep(rcvIntervalTimeLeft);
  29. rcvIntervalTimeLeft = 0;
  30. }
  31. } catch (InterruptedException e) {
  32. e.printStackTrace();
  33. }
  34. }
  35. // MQ Subscribe interval logical ends
  36. System.out.println("Receive: " + message);
  37. /*
  38. * Put your business logic here.
  39. */
  40. doSomething();
  41. return Action.CommitMessage;
  42. }
  43. });
  44. consumer.start();

Running result

Run the standalone Consumer for consumption. Assuming the message queue always has more messages than you can consume, the testing will be conducted in three stages, each lasting for five minutes. We can achieve the following results by using ACM configuration push service.

  • RCV_INTERVAL_TIME = 100 ms
  • RCV_INTERVAL_TIME = 5000 ms
  • RCV_INTERVAL_TIME = 1000 ms

For a testing case with a standalone MQ consumption business processing time of approximately 100 ms and a standalone concurrent thread number of 20, the testing results are as follows:

  • RCV_INTERVAL_TIME = 100 ms: The average consumption performance is approximately 9000 tpm
  • RCV_INTERVAL_TIME = 5000 ms: The average consumption performance is restricted to approximately 200 tpm
  • RCV_INTERVAL_TIME = 1000 ms: The average consumption performance rises to approximately 1100 tpm

As expected, these results indicate that the consumption is inversely proportional to tpm. Most importantly, the application is not interrupted during the whole process. The traffic control push takes effect on distributed clusters in less than one second. The standalone performance result is as follows:


Thank you! We've received your feedback.