QMQ消息中间件安装与使用指南

2026-01-17 09:33:00作者：庞队千Virginia

一、项目介绍

QMQ (Qunar Message Queue) 是一款由去哪儿网研发并广泛应用在其内部业务中的消息中间件。自2012年以来，QMQ在去哪儿网的所有业务场景中扮演了核心角色，不仅应用于与交易密切相关的订单管理，还服务于诸如价格查询及搜索等高并发需求场景。该系统以其低延迟（平均端到端延迟控制在10毫秒以内）、高性能（日常消息处理峰值约为每秒60万条）以及强大的可扩展性（支撑近4万个消息主题），成为企业级消息传递解决方案的理想选择。

QMQ主要功能亮点包括：

异步实时消息：即时发送与接收。
延迟/定时消息：精确到秒级的时间调度。
广播消息：确保每个消费者接收到一致的信息，适用于缓存更新等场景。
服务端与客户端过滤机制：基于标签（Tag）实现高效数据筛选。
幂等性保证：多次消费同一消息不会产生重复操作。
丰富的监控与追踪：集成OpenTracing标准，便于故障排查。
事务消息处理：保障数据一致性。
灵活的伸缩性：消费者与服务器节点可根据实际负载动态调整。
语言多样性：支持Java、.NET客户端，兼容多种应用场景。

二、项目快速启动

为了迅速让QMQ运行起来，在你的开发环境中搭建一个基本的QMQ环境至关重要。以下步骤将指导你完成这一过程。

环境准备

JDK: 对于客户端，至少需1.7版本；对于服务器，至少需1.8版本。
Maven: 用于自动化构建与依赖管理。

Maven依赖添加

要在你的项目中使用QMQ，首先需要将QMQ客户端库添加至你的Maven pom.xml文件中：

<dependencies>
    <!-- 添加QMQ客户端依赖 -->
    <dependency>
        <groupId>com.qunar.qmq</groupId>
        <artifactId>qmq</artifactId>
        <version>指定版本号</version>
    </dependency>
</dependencies>

请注意替换上述示例中的指定版本号为你所需的QMQ具体版本号。

发送与接收消息示例

初始化Producer

import com.qunar.qmq.client.Message;
import com.qunar.qmq.client.Producer;

public class SendMessageExample {
    
    public static void main(String[] args) throws Exception {
        
        // 创建producer实例
        Producer producer = new Producer("YOUR_PRODUCER_GROUP");
        producer.start();
        
        // 准备要发送的消息
        Message msg = new Message("YOUR_TOPIC_NAME", "Hello QMQ".getBytes());
        
        // 发送消息
        SendResult sendResult = producer.send(msg);
        
        // 输出发送结果
        System.out.println(sendResult.getMsgId());
        
        // 关闭producer
        producer.shutdown();
    }
}

初始化Consumer

import com.qunar.qmq.client.Consumer;
import com.qunar.qmq.client.MessageListener;
import com.qunar.qmq.client.MessageQueue;

public class ReceiveMessageExample {

    public static void main(String[] args) throws Exception {
        
        // 创建consumer实例
        Consumer consumer = new Consumer("YOUR_CONSUMER_GROUP");
        consumer.subscribe("YOUR_TOPIC_NAME");
        consumer.registerMessageListener(new MessageListener() {
            @Override
            public boolean consumeMessage(List<MessageExt> msgs) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return true;
            }
        });
        
        // 启动consumer
        consumer.start();
        
        // 保持线程阻塞以便持续监听消息
        Thread.sleep(Long.MAX_VALUE);
    }
}