QMQ: Design, Usage, and Implementation of Qunar's Distributed Message Queue

In 2012 Qunar began a service‑oriented transformation and needed an asynchronous communication mechanism beyond RPC; QMQ was created as an internal distributed message queue to provide decoupling, eventual consistency, flow control and high reliability.

The first version of QMQ stored messages in MySQL and Redis and later evolved a new storage model inspired by Kafka and RocketMQ, adding a custom consumer management layer to address throughput, backlog handling and push‑model limitations.

Key concepts include producer (message sender), consumer (message receiver), broker (server), subject (topic), consumer group (shared consumption state) and broadcast consumption where each consumer receives all messages.

Sending messages – create a producer, set the app code and meta‑server address, initialize it, generate a Message , set key/value properties and call sendMessage . A callback can be provided via MessageSendStateListener to get success or failure notifications.

MessageProducerProvider producer = new MessageProducerProvider();
producer.setAppCode("your app");
producer.setMetaServer("http://
/meta/address");
producer.init();
Message message = producer.generateMessage("qmq_subject");
message.setProperty("key", "value");
producer.sendMessage(message);

For delayed messages, set the delay time before sending:

Message message = producer.generateMessage("qmq_delay_subject");
message.setProperty("key", "value");
message.setDelayTime(15, TimeUnit.MINUTES);
producer.sendMessage(message);

Consuming messages – QMQ can be integrated with Spring either via XML configuration:

<qmq:consumer appCode="your app" metaServer="http://meta server/meta/address"/>
<bean id="qmqExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean">
    <property name="corePoolSize" value="2"/>
    <property name="maxPoolSize" value="2"/>
    <property name="queueCapacity" value="1000"/>
    <property name="threadNamePrefix" value="qmq-process"/>
</bean>

or via Spring Boot annotations:

@Configuration
@EnableQmq(appCode="your app", metaServer="http://
/meta/address")
public class Config {}

@QmqConsumer(subject="qmq_subject", consumerGroup="group", executor="executor bean name")
public void onMessage(Message message) {
    String value = message.getStringProperty("key");
    // process message
}

Consumer options include consumeMostOnce=true for at‑most‑once delivery and isBroadcast=true for broadcast consumption.

Transactional messages – By configuring a TransactionProvider (e.g., SpringTransactionProvider ) and using Spring's @Transactional , message sending can be part of the same database transaction as business operations, guaranteeing strong consistency. A watchdog service retries failed sends after commit.

<bean id="transactionProvider" class="qunar.tc.qmq.producer.tx.spring.SpringTransactionProvider">
    <constructor-arg name="bizDataSource" ref="dataSource"/>
</bean>

<bean id="messageProducer" class="qunar.tc.qmq.producer.MessageProducerProvider">
    <property name="appCode" value="your app"/>
    <property name="metaServer" value="http://
/meta/address"/>
    <property name="transactionProvider" ref="transactionProvider"/>
</bean>

Within a transactional method:

@Transactional
public void pay(Order order) {
    PayTransaction t = buildPayTransaction(order);
    payDao.append(t);
    producer.sendMessage(buildMessage(t));
}

Overall architecture – QMQ consists of three core components: metaserver (stores metadata such as subject routing and cluster membership), broker (real‑time message handling) and delay (delayed message handling). Each cluster contains multiple groups with master‑slave nodes, and heartbeats keep the system healthy.

Real‑time queue design – QMQ separates consumption from storage using three logs: a message log (append‑only storage of all subjects), a consume log (index of a subject’s messages) and a pull log (records each consumer’s pull sequence, serving as the consumption offset). This decouples consumer scaling from partitioning.

Delayed queue design – Delayed messages use a two‑layer hash‑wheel timer. The first layer is a disk‑based schedule log with one file per hour (supporting up to two years, ~17 500 files). The second layer is an in‑memory wheel that loads the upcoming hour’s indexes. A dispatch log records which delayed messages have been delivered to survive restarts.