RocketMQ 和 RabbitMQ 的比較以及 RocketMQ 的使用

作者：oec2003 2025-01-02 08:31:33

消息隊列在項目中會經常用到，目前我們使用的是 RabbitMQ，但在 Java 技術棧下，RocketMQ 使用的比較多。下面比較下 RabbitMQ 和 RocketMQ。

RabbitMQ 和 RocketMQ 對比

1.設計理念和架構

RabbitMQ：

基于 AMQP（Advanced Message Queuing Protocol）協議，使用 Erlang 語言開發。Erlang 的天生高并發和容錯性使得 RabbitMQ 在穩定性方面表現出色。RabbitMQ 的核心概念是 Exchange（交換機）和 Queue（隊列），消息通過 Exchange 路由到 Queue，再由消費者消費。這種模型非常靈活，支持多種消息路由模式。

RocketMQ：

源于阿里巴巴，后捐獻給 Apache 基金會，所以現在的官網是：https://rocketmq.apache.org/ 。使用 Java 語言開發，更貼近 Java 技術棧。RocketMQ 的核心概念是 Topic（主題），消息發送到 Topic，消費者訂閱 Topic 進行消費。RocketMQ 的設計目標是高吞吐量、低延遲和高可靠性，適合大規模分布式系統。

RocketMQ 的設計理念更偏向于解決互聯網場景下的具體問題，如海量消息處理、消息順序性等。

2.性能

吞吐量：RocketMQ 在吞吐量方面通常優于 RabbitMQ，尤其是在高并發場景下。RocketMQ 的設計更偏向于高吞吐的消息傳遞，而 RabbitMQ 更注重消息的可靠性和靈活性。

延遲：兩者在延遲方面都表現不錯，但在極端高負載情況下，RocketMQ 的延遲可能更低一些。

不過在 ToB 的一些業務場景，RabbitMQ 是可以勝任的。

3.特性

消息路由：RabbitMQ 支持多種 Exchange 類型（Direct、Topic、Fanout、Headers），提供更豐富的消息路由策略。RocketMQ 主要使用 Topic 進行消息路由，相對簡單。

消息過濾：RocketMQ 支持基于 Tag 和 SQL 的消息過濾，方便消費者按需訂閱消息。RabbitMQ 的消息過濾相對較弱。

事務消息：RocketMQ 提供了分布式事務消息的支持，可以保證消息生產和本地事務的原子性。RabbitMQ 沒有直接提供事務消息的支持，需要通過其他方式實現。

延遲消息：RocketMQ 支持延遲消息，可以實現定時任務等功能。RabbitMQ 通過插件可以實現類似功能。

監控和管理：RocketMQ 和 RabbitMQ 都提供了豐富的監控指標和管理工具，相比之下我更喜歡 RocketMQ 的管理工具。

4.創新點

RabbitMQ:

插件系統設計靈活，易于擴展
虛擬主機（vhost）概念，實現多租戶隔離
內存和磁盤節點的混合部署方案

RocketMQ:

基于文件的消息存儲系統，避免了緩存未刷盤導致的消息丟失
Pull 模式和長輪詢機制的結合，平衡了實時性和性能
消息過濾支持在 Broker 端完成，減少網絡傳輸開銷

5.Exchange 和 Topic 的區別

RabbitMQ的 Exchange 和 RocketMQ 的 Topic 在消息路由機制上有以下主要區別：

概念和角色

RabbitMQ Exchange 是一個路由組件，負責接收生產者發送的消息并將其路由到一個或多個隊列，作為消息的"交換機"，它不存儲消息，只負責消息的路由轉發，需要通過 binding key 與 Queue 建立綁定關系。

RocketMQ Topic 是消息的邏輯分類，直接作為消息的存儲和投遞單元，包含多個消息隊列（MessageQueue），用于存儲消息，消費者直接訂閱 Topic 即可接收消息。

RabbitMQ 的 key 綁定和 Exchange、隊列的關系，一開始不太容易理解，相比之下 RocketMQ 的 Topic 和隊列關系更清晰。

路由方式

RabbitMQ Exchange 支持四種路由策略，路由更加靈活，可以實現復雜的消息分發模式

Direct：根據routing key精確匹配
Topic：根據routing key的模式匹配
Fanout：廣播到所有綁定的隊列
Headers：根據消息屬性匹配

RocketMQ Topic 采用發布/訂閱模式，更加簡單直接，通過Tag機制實現消息過濾。支持消息隊列的負載均衡。

消息存儲

RabbitMQ Exchange 不存儲消息，消息存儲在 Queue 中，消息一旦被路由到 Queue 就與 Exchange 無關。

RocketMQ Topic 直接存儲消息，每個 Topic 包含多個消息隊列。消息存儲在 CommitLog 中，通過 ConsumeQueue 建立索引。

Docker-compose 部署 RocketMQ

同樣是使用容器進行部署，RabbitMQ 一個容器搞定，RocketMQ 需要兩個容器（NameServer 和 Broker），如果需要 web 管理工具，還需要再單獨部署一個容器。

當進行集群模式部署時，RocketMQ 的下載包中有各種集群模式的示例配置文件，這對新手非常友好。

下面是部署 RocketMQ 的 docker-comopose 文件的內容：

version: '3'

# 定義自定義網絡
networks:
  rmq_network:
    driver: bridge
    ipam:
      config:
        - subnet: 192.168.10.0/24

services:
  # RocketMQ Name Server
  namesrv:
    image: apache/rocketmq:5.1.4
    container_name: rmqnamesrv
    networks:
      rmq_network:
        ipv4_address: 192.168.10.2
    ports:
      - 9876:9876
    volumes:
      - ./data/namesrv/logs:/home/rocketmq/logs
    command: sh mqnamesrv
    environment:
      - JAVA_OPT_EXT=-server -Xms512m -Xmx512m

  # RocketMQ Broker
  broker:
    image: apache/rocketmq:5.3.1
    container_name: rmqbroker
    networks:
      rmq_network:
        ipv4_address: 192.168.10.3
    ports:
      - 10909:10909
      - 10911:10911
      - 10912:10912
    volumes:
      - ./data/broker/logs:/home/rocketmq/logs
      - ./data/broker/store:/home/rocketmq/store
      - ./conf/broker.conf:/home/rocketmq/conf/broker.conf
    command: sh mqbroker -c /home/rocketmq/conf/broker.conf
    environment:
      - JAVA_OPT_EXT=-server -Xms512m -Xmx512m
    depends_on:
      - namesrv

  # RocketMQ Dashboard 
  dashboard:
    image: apacherocketmq/rocketmq-dashboard:1.0.0
    container_name: rmqdashboard
    networks:
      rmq_network:
        ipv4_address: 192.168.10.4
    ports:
      - 19080:8080
    environment:
      - JAVA_OPTS=-Drocketmq.namesrv.addr=namesrv:9876
    depends_on:
      - namesrv

broker.conf 的內容如下：

# broker集群名稱
brokerClusterName = DefaultCluster
# broker名稱
brokerName = broker-a
# broker編號，0表示master，大于0表示slave
brokerId = 0
# 刪除過期文件時間點，默認是凌晨4點
deleteWhen = 04
# 文件保留時間，默認48小時
fileReservedTime = 48
# broker角色，ASYNC_MASTER=異步復制Master，SYNC_MASTER=同步雙寫Master，SLAVE=slave節點
brokerRole = ASYNC_MASTER
# 刷盤方式，ASYNC_FLUSH=異步刷盤，SYNC_FLUSH=同步刷盤
flushDiskType = ASYNC_FLUSH
# nameServer地址，分號分割
namesrvAddr = namesrv:9876
# 在發送消息時，自動創建服務器不存在的topic，默認創建的隊列數
defaultTopicQueueNums = 4
# 是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉
autoCreateTopicEnable = true
# 是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉
autoCreateSubscriptionGroup = true
# brokerIP1 注意：本地測試使用本機的宿主機的IP
brokerIP1=192.168.1.109

代碼示例

對于消息隊列，單播、廣播、重試，這三種場景用的比較多。下面就看看這三個場景是怎么實現的。

創建生產者 Service 類來處理消息的發送：

@Slf4j
@Service
public class MessageProducerService {

    // RocketMQ消息主題
    public static final String TOPIC_UNICAST = "topic-unicast";
    public static final String TOPIC_BROADCAST = "topic-broadcast";
    public static final String TOPIC_RETRY = "topic-retry";

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 發送單播消息（點對點）
     * 單播消息會被消費組中的某一個消費者消費
     */
    public void sendUnicastMessage(MessageEvent message) {
        rocketMQTemplate.convertAndSend(TOPIC_UNICAST, message);
        log.info("Unicast message sent: {}", message);
    }

    /**
     * 發送廣播消息
     * 廣播消息會被所有訂閱該主題的消費者消費
     */
    public void sendBroadcastMessage(MessageEvent message) {
        rocketMQTemplate.convertAndSend(TOPIC_BROADCAST, message);
        log.info("Broadcast message sent: {}", message);
    }

    /**
     * 發送需要重試的消息
     * 使用異步發送方式，并在回調中處理發送結果
     */
    public void sendRetryMessage(MessageEvent message) {
        rocketMQTemplate.asyncSend(TOPIC_RETRY, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("Retry message sent successfully: {}, result: {}", message, sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("Failed to send retry message: {}, error: {}", message, throwable.getMessage());
            }
        });
    }
}

創建消費者 Service 類來處理消息的接收：

@Slf4j
@Service
public class MessageConsumerService {

    // RocketMQ消息主題
    public static final String TOPIC_UNICAST = "topic-unicast";
    public static final String TOPIC_BROADCAST = "topic-broadcast";
    public static final String TOPIC_RETRY = "topic-retry";
    /**
     * 單播消息消費者
     * consumeMode默認為CONCURRENTLY（并發消費）
     */
    @Service
    @RocketMQMessageListener(
            topic = TOPIC_UNICAST,
            consumerGroup = "unicast-consumer-group"
    )
    public class UnicastMessageListener implements RocketMQListener<MessageEvent> {
        @Override
        public void onMessage(MessageEvent message) {

            log.info("Received unicast message: {}", message);
        }
    }

    /**
     * 廣播消息消費者
     * messageModel設置為BROADCASTING表示廣播模式
     */
    @Service
    @RocketMQMessageListener(
            topic = TOPIC_BROADCAST,
            consumerGroup = "broadcast-consumer-group",
            messageModel = MessageModel.BROADCASTING
    )
    public class BroadcastMessageListener implements RocketMQListener<MessageEvent> {
        @Override
        public void onMessage(MessageEvent message) {
            log.info("Received broadcast message: {}", message);
        }
    }

    /**
     * 重試消息消費者
     * 配置了重試次數和重試間隔
     */
    @Service
    @RocketMQMessageListener(
            topic = TOPIC_RETRY,
            consumerGroup = "retry-consumer-group",
            maxReconsumeTimes = 3,    // 最大重試次數
            delayLevelWhenNextConsume = 2  // 重試間隔級別
    )
    public class RetryMessageListener implements RocketMQListener<MessageEvent> {
        @Override
        public void onMessage(MessageEvent message) {
            try {
                // 模擬處理失敗的情況
                if (message.getContent().contains("error")) {
                    throw new RuntimeException("Processing failed, will retry");
                }
                log.info("Received retry message: {}", message);
            } catch (Exception e) {
                log.error("Error processing message: {}, error: {}", message, e.getMessage());
                throw e; // 拋出異常觸發重試機制
            }
        }
    }
}

創建 MessageController 來進行測試：

@RestController
@RequestMapping("/api/messages")
public class MessageController {

    @Autowired
    private MessageProducerService producerService;

    @PostMapping("/unicast")
    public String sendUnicastMessage(@RequestParam String content) {
        MessageEvent message = new MessageEvent()
                .setId(UUID.randomUUID().toString())
                .setContent(content)
                .setTimestamp(System.currentTimeMillis());
        producerService.sendUnicastMessage(message);
        return "Unicast message sent successfully";
    }

    @PostMapping("/broadcast")
    public String sendBroadcastMessage(@RequestParam String content) {
        MessageEvent message = new MessageEvent()
                .setId(UUID.randomUUID().toString())
                .setContent(content)
                .setTimestamp(System.currentTimeMillis());
        producerService.sendBroadcastMessage(message);
        return "Broadcast message sent successfully";
    }

    @PostMapping("/retry")
    public String sendRetryMessage(@RequestParam String content) {
        MessageEvent message = new MessageEvent()
                .setId(UUID.randomUUID().toString())
                .setContent(content)
                .setTimestamp(System.currentTimeMillis());
        producerService.sendRetryMessage(message);
        return "Retry message sent successfully";
    }
}

責任編輯：姜華來源：不止dotNET