From Alibaba to Apache: RocketMQ’s Past, Present, and Future

As the idea of open source prevails these years, the leading technology companies continue to share their own research projects with the open source community. In 2012, Alibaba open-sourced its RocketMQ, a third-generation distributed messaging middleware. Through several years of technical improvement, RocketMQ is now capable of transferring trillions of concurrent online messages as of Alibaba's Nov. 11th Shopping Festival.

In November 2016, Alibaba donated RocketMQ to the Apache Software Foundation (ASF) as an incubator project. That was a huge step for Alibaba to make it through ASF's competitive evaluation process. It well supported that Chinese developers became increasingly influential in the global open source community against the past stereotype.

On February 20th, 2017, Apache RocketMQ released its 4.0.0 version. The experts claimed that the new version would start a new generation of programming model in the fields of e-commerce, finance, big-data, and the internet of things.

What is the technical core of RocketMQ, and why did it stand out in front of ASF? What are the implications behind its successful donation? You will find the answers in this article.

Three generations of evolution

RocketMQ's evolution can be divided into three stages.

The first generation adopts the push mode in which data are stored in relational database. Low latency message delivery and distributed transactions meet Alibaba's demands and contribute to RocketMQ's wide adoption.

The second generation adopts the pull mode in which data storage is managed by file systems. Its performance on log collection is comparable to Kafka. Taking account of Alibaba's application scenarios, especially the requirement for high reliability during transaction processes, the second-generation RocketMQ does not blindly seek the throughput but also stresses stability and reliability. The long pulling mechanism makes the response time of the system no slower than that in the first generation.

The third generation, RocketMQ mainly uses the Pull mode along with some Push operations. Largely optimized, it has low latency and high performance, especially in high concurrency and massive amounts of data scenarios. Since 2012, RocketMQ has challenged the task of supporting Alibaba's annual Nov. 11th Global Shopping Festival multiple times. During the 2016 Festival, RocketMQ robustly provided stable infrastructure with a transfer throughput of more than one trillion messages.

Technology Overview

The greatest innovation of RocketMQ is high throughput, high reliability, and low latency, supporting massive messages through exquisite scale-out and scale-up. RocketMQ consists of four parts, name servers, brokers, producers and consumers, as shown in screenshot below. Each of them can be horizontally extended without single point of failures.

(Click on the image to enlarge it)

Name servers provide lightweight service discovery and routing. Each name server records full routing information, provides equivalent reading and writing service, and supports fast storage expansion.

Brokers take care of message storage by providing lightweight queue mechanisms with which messages are grouped by topic. It supports the push and pull modes, fault tolerance mechanism (2 copies or 3 copies), strong padding of peaks and capacity of accumulating hundreds of billion messages in their original time order. In addition, Brokers provide disaster recovery, rich metrics statistics, and alert mechanisms, all of which are lack in traditional messaging systems.

Producers support distributed deployment. Distributed producers send messages to the broker cluster through multiple load balancing modes. The sending processes support fast failure and have low latency.

Consumers support distributed deployment in the push and pull models as well. It also supports cluster consumption and message broadcast. It provides real-time message subscription mechanism that meets most consumer scenarios.

RocketMQ's website provides a quick-start guide for those who want a try.

Behind-the-scene Hero for the Festival

Similar to the Black Friday, the annual Global Shopping Festivals are discount celebrations on Alibaba's online shopping platforms. To prepare for the 2016 Shopping Festival, RocketMQ team mainly focused on two goals, to optimize slow requests and to enhance the unified storage engine.

The first goal is to deal with slow requests, a long tail latency problem due to the heavy load of messages, by optimizing the entire cluster long-tail latency. That was a huge challenge. After more than one month of experiments, it was achieved in the Festival that, 99.996% of the delays were within 10 ms, and 99.6% of the delays were within 1 ms. The team mainly focused on RocketMQ storage layer algorithm optimization, and JVM and operating system tuning.

The second goal is to enhance the unified storage engine, which mainly dealt with high availability and low cost of the message engine. The team comprehensively replaced and transplanted the storage module of Notify under the multi-generation message engine coexistence.

With these optimizations, during the 2016 global Nov. 11th Shopping Festival, the Alibaba Group produced 1.2 trillion message transmissions, almost twice as much as of 2015. During the peak hours, the throughput of the message production was around 20,000,000 per second while the message consumption was 15,000,000 per second.

The following screenshot visualizes the RocketMQ behavior in throughput and latency on a machine with the following configurations:

Intel(R) Xeon(R) CPU E5-2682 v4 @ 2.50GHz with 64 cores
Memory 189GB
SSD 1.8TB
NICs 2000Mb/s
kernel 2.6.32-x86_64

(Click on the image to enlarge it)

RocketMQ vs. Other Messaging Solutions

We compare RocketMQ with ZeroMQ, RabbitMQ, ActiveMQ and Kafka from three aspects: technical architecture, protocol support and application scenario.

Whether they have the client server architecture

Among similar products, Let's briefly look at ZeroMQ firstly. Named as 0MQ, it is more like an embedded network class library, a communication component that focus on the transports layer rather than a MQ with a traditional client server (CS) architecture.

Which Specification / Protocol they implement

Now, let's view RabbitMQ, ActiveMQ, Kafka and RocketMQ from the design perspective. A reference implementation of AMQP specification, RabbitMQ is a realization of AMQP, which is an exhaustive, highly systematic, and slightly complicated wire layer protocol. Currently, RabbitMQ is also a messaging service chosen by the OpenStack Iaas platform, which indicates the strength of its supporters.

ActiveMQ, originally donated to Apache from LogicBlaze, is a reference implementation of JMS specification and a well-known messaging engine in Apache. Although JMS is an API-level specification, it defines some implementation constraints and does not support multiple languages. ActiveMQ has an active ecology and many sub-projects under the top project in Apache, including Artemis evolved, which come from HornetQ, and Apollo, which is based on Scala's next-generation AMQ.

Where they are adopted

Kafka, originally designed for log processing, is a large data channel: it has the pursuit of high throughput while possible to lose messages. R&D team in charge of the project, carried out commercial packaging for Kafka. It is now widely adopted in many small and medium sized companies around the world.

RocketMQ, born of the finance industry and Internet, seeks high reliability, high availability, high concurrency, and low latency. Nowadays, RocketMQ is not only used within the Alibaba Group but also by hundreds of companies, and research and educational institutions. The ecosystem projects around it can be found here.

Three Keys as a Distributed Messaging Engine

As a distributed messaging engine, RocketMQ's keys are keeping order of messages, checking for duplicates, and following CAP principles.

(I) Order of messages

Keeping order of messages is an appealing feature of RocketMQ. It only preserves global ordering when each message can be marked by a unique hashable tag, such as seller accounts and order numbers. The code snippet like the following:

producer code

//This is a sharding key for global message order, it can be seller accounts or order numbers as described before.
int orderId;
…
//You can assign which queue you want to put
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);

consumer code

//MessageListenerOrderly is designed for consuming message orderly, you don't have to care about the backend concurrency details.
consumer.registerMessageListener(new MessageListenerOrderly() {
            AtomicLong consumeTimes = new AtomicLong(0);
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(false);
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    return ConsumeOrderlyStatus.SUCCESS;
                } else if ((this.consumeTimes.get() % 3) == 0) {
                    return ConsumeOrderlyStatus.ROLLBACK;
                } else if ((this.consumeTimes.get() % 4) == 0) {
                    return ConsumeOrderlyStatus.COMMIT;
                } else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

More examples to preserve message ordering can be found here.

The technique behind is relatively simple, to ensure a single instance of channel operations, such as single-process, single-threaded writing and reading. ActiveMQ's feature, Exclusive Consumer, operates in a similar way.

As one can see, RocketMQ's realization compromises throughput and brings another problem - hot issues. For example, on the day of Nov. 11th, the simple hashing results in massive trading requests congesting in a single channel. The problem still exists in the latest RocketMQ version but it is expected to be polished this year.

(II) Removal of duplicates

The field has a Qos classification of message delivery: at most once, at least once, and exactly once. Almost all of the MQ products claim their feature of delivering at least once.

Message duplicates are inevitable, especially in a distributed network. The shortcoming can be viewed as part of the TCP protocol, such as failure to retransmit. Businesses are often sensitive to message duplication. The industry usually use global storage to eliminate duplicates, such as Artmis. Currently, RocketMQ does not check for message duplication, leading users to build or buy their own external global storage for re-processing. The next generation of RocketMQ, however, will support this feature. We hope to support several Qos of duplicate message detection, eliminating duplicates in various timeline.

(III) A Tour of CAP Theorem

Distributed systems are defined as a series of decentralized autonomous components through concurrent collaboration in the internet that form a coherent software system. Its features include resource sharing, parallel concurrent, and fault tolerance. CAP, BASE, Paxos, Transaction together builds up the fundamental theories of the distributed systems.

Let us have a look at the CAP theorem. CAP stands for consistency, availability, and partition tolerance. The proposer of the theorem, Eric Brewer, defines consistency as the principle that a service that is consistent communicating fully or not at all. Please note that, consistency here is different than that in the ACID attribute of the database, which refers to the integrity constraints between data operations, such as foreign key constraints. In a distributed environment, consistency is a multi-node where data can exist in multiple states of replication. Service availability can be categorized as writable and readable. In a distributed environment, it refers to a system reading and writing operations within a time frame. Partition tolerance, as it is, no set of Failures less than total network failure is allowed to cause the system to respond incorrectly.

The CAP theorem explores the balance between consistency and usability under conditions with or without storage partition.

When no partition: both C and A are fulfilled as well as full ACID transaction support. Can compromise C at some degree for better performance and scalability.

When has partition: Prefer A (focus on district recovery), it requires solutions for partition in the state of before the start, in progress, and after the recovery, and appropriate compensation processing mechanism.

RocketMQ pursues more A and P. Even most powerful systems have storage limits, sufficient storage is a prerequisite of availability. RocketMQ designs a series of mechanism for downgrading, flow limit and fuse mechanism, to ensure the availability under heavier loads. Dealing with frequent transactions, the C and P mechanism for RocketMQ can compromise system availability for data consistency while satisfying the partition fault tolerance of distributed systems. For the technical realization, RocketMQ follows the Zab protocol using a distributed lock and notification mechanism for the consistency among multiple copies of data.

Open Source and Community Maintenance

RocketMQ has been open-source since 2012. Though several years of technical improvement, after careful considerations, several discussions and approvals, the RocketMQ team decided to donate the project to Apache.

The team also produces a commercial distribution called Aliware MQ. It supports the TCP, HTTP and MQTT protocols, enhances the maintenance tools of ecological integration (including visualization of the trajectory, resource reporting statistics monitoring and alerts, as well as Kafka integration). Designed for enterprise, it has multi-region deployment with high availability disaster recovery characteristics on the public cloud and private cloud.

As for community maintenance, the team has adopted a strategy similar to Apache's other top-level projects as the team devotes itself to the high quality of its products. Since the beginning, the team conducts the version planning through a series of standardized software development processes, such as milestone discussions, features design, code testing, peer reviews, integration tests, release discussions, and release announcements. Whatsmore, the team interacts with the open source community through activities like line Meetups, Workshops, ApacheCon and Code Marathon, to attract new contributors and committers.

Future Plans

The team hopes the fourth generation of RocketMQ to support a set of vendor-independent API layer specification. The team has contacted technical leaders from Twitter, Yahoo and some other companies, to jointly draft and polish the specification. RocketMQ will be among the first group of projects to implement the specification. The team welcomes MQ vendors and distributed enthusiasts to join.

The team has just released the first version of the fourth generation a few days ago, also RocketMQ's first release after joining the Apache community. The team will upgrade and re-organize the entire engine in May 2017. Welcome to use RocketMQ, provide feedback, and make contributions!

You are welcomed to check more information about the project on Apache RocketMQ official website, the Alibaba Cloud community, the official middleware blog and the Alibaba's e-book on RocketMQ.

About the Authors

Wang Xiaorui is Alibaba's middleware messaging chief architect. He has led Alibaba's mission of optimizing the engine for the Nov. 11th Global Shopping Festival many times. He also has rich speaker experience around the world. Such as the topic "Alibaba Open Source Culture" in the 10th Northeast Asia OSS Promotion Forum Summit in Tokyo. The topic "Trillion Distributed Messaging - Apache RocketMQ" QCon Beijing. His contact email is here.

Feng Jia, Alibaba's middleware architect, has abundant experience in distributed software architecture, high concurrency website design and performance tuning. An open source enthusiast, his interests include Hbase / Hadoop / Spark / Flink and other big-data technology stack. He is in charge of the Alibaba messaging ecosphere spread and cloud commercialization solution product - Aliware MQ. He is an active tech. speaker in China, also the speaker at ApacheCon North America 2017. His contact email is here.