快捷導(dǎo)航

RocketMQ保證消息的有序性的案例分享

更新時(shí)間：2024年04月25日 08:50:22 作者：一只愛擼貓的程序猿

Apache RocketMQ 是一個(gè)常用的開源消息中間件,它提供了強(qiáng)大的有序消息處理能力,這里我們會(huì)探討 RocketMQ 是如何保證消息的有序性的,包括其設(shè)計(jì)原理和相關(guān)的源碼實(shí)現(xiàn),需要的朋友可以參考下

在分布式系統(tǒng)中，消息隊(duì)列（MQ）的有序性是一個(gè)重要的特性，尤其是在需要保證事件順序執(zhí)行的業(yè)務(wù)場(chǎng)景下。Apache RocketMQ 是一個(gè)常用的開源消息中間件，它提供了強(qiáng)大的有序消息處理能力。這里我們會(huì)探討 RocketMQ 是如何保證消息的有序性的，包括其設(shè)計(jì)原理和相關(guān)的源碼實(shí)現(xiàn)。

簡單原理

RocketMQ 有序消息的基本概念

RocketMQ 保證有序性的主要方法是通過順序消息來實(shí)現(xiàn)的。在 RocketMQ 中，順序消息分為全局順序和分區(qū)順序兩種：

全局順序：指的是消息全局范圍內(nèi)的有序，也就是在所有的消息中，都是按照發(fā)送的順序來消費(fèi)。
分區(qū)順序：指的是在同一個(gè)隊(duì)列（Queue）中的消息是有序的，而不同隊(duì)列間的消息并不保證有序。

RocketMQ 默認(rèn)使用分區(qū)順序，通過將同一個(gè) topic 下的消息分到同一個(gè)隊(duì)列（queue）中，來保證隊(duì)列內(nèi)的消息有序。

RocketMQ 有序消息的實(shí)現(xiàn)機(jī)制

消息發(fā)送

在發(fā)送端，RocketMQ 通過確保生產(chǎn)者向同一個(gè)隊(duì)列（Queue）發(fā)送消息來保證消息的有序性。生產(chǎn)者在發(fā)送消息時(shí)可以指定消息的 keys 或者其他屬性，RocketMQ 通過這些屬性計(jì)算消息應(yīng)該發(fā)送到哪個(gè)隊(duì)列。

源碼示例（偽代碼）：

public class Producer {
    public void sendMessages(List<Message> messages) {
        for (Message msg : messages) {
            int queueId = this.calculateQueueId(msg);
            msg.setQueueId(queueId);
            this.sendMessageToQueue(msg, queueId);
        }
    }

    private int calculateQueueId(Message msg) {
        // 使用 hash 算法基于 message key 計(jì)算隊(duì)列 ID
        return Math.abs(msg.getKey().hashCode()) % this.queueSize;
    }
}

消息消費(fèi)

在消費(fèi)端，RocketMQ 使用單線程消費(fèi)模式來保證同一個(gè)隊(duì)列的消息順序性。消費(fèi)者會(huì)固定分配到某個(gè)隊(duì)列，而且是單線程從該隊(duì)列拉取并處理消息，從而保證消息的有序處理。

源碼示例（偽代碼）：

public class Consumer {
    public void consume() {
        while (true) {
            Message msg = this.pullMessage();
            this.processMessage(msg);
        }
    }
}

簡單案例

在Spring Boot中使用RocketMQ來保證消息的隊(duì)列順序性，我們需要配置RocketMQ的客戶端和服務(wù)器端以支持順序消息。以下是一個(gè)基于RocketMQ和Spring Boot實(shí)現(xiàn)的消息順序發(fā)送和消費(fèi)的例子。這個(gè)場(chǎng)景假設(shè)我們需要在一個(gè)電商系統(tǒng)中處理訂單狀態(tài)更新，訂單狀態(tài)更新必須按照順序來處理，以避免狀態(tài)不一致。

步驟1: 添加依賴

首先，確保你的pom.xml中加入了RocketMQ的Spring Boot Starter依賴。

<dependencies>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.1.0</version>
    </dependency>
</dependencies>

步驟2: 配置RocketMQ

在application.yml或application.properties中配置RocketMQ的基礎(chǔ)屬性：

rocketmq:
  name-server: 127.0.0.1:9876  # 修改為你的NameServer地址
  producer:
    group: order_producer_group
    send-message-timeout: 3000
  consumer:
    group: order_consumer_group
    consume-thread-min: 1
    consume-thread-max: 1

步驟3: 生產(chǎn)者配置

創(chuàng)建一個(gè)生產(chǎn)者服務(wù)，這個(gè)服務(wù)將訂單狀態(tài)更新作為順序消息發(fā)送。

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderStatusProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendOrderStatusUpdate(String orderStatus, String orderId) {
        // 使用訂單ID作為key保證同一個(gè)訂單的更新在同一個(gè)隊(duì)列
        SendResult result = rocketMQTemplate.syncSendOrderly("order-topic", orderStatus, orderId);
        System.out.println("Message sent, result: " + result.getSendStatus());
    }
}

步驟4: 消費(fèi)者配置

創(chuàng)建一個(gè)消費(fèi)者服務(wù)，這個(gè)服務(wù)將按順序消費(fèi)訂單狀態(tài)更新。

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order_consumer_group", consumeMode = ConsumeMode.ORDERLY)
public class OrderStatusConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("Received order update: " + message);
        // 處理訂單更新邏輯
        processOrderUpdate(message);
    }

    private void processOrderUpdate(String status) {
        // 實(shí)現(xiàn)訂單更新處理邏輯
        System.out.println("Processing order status update: " + status);
    }
}

步驟5: 測(cè)試消息順序

你可以通過編寫一個(gè)簡單的測(cè)試來發(fā)送多個(gè)消息，并觀察消費(fèi)者是否按順序接收它們。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class OrderStatusTestRunner implements CommandLineRunner {

    @Autowired
    private OrderStatusProducer producer;

    @Override
    public void run(String... args) throws Exception {
        producer.sendOrderStatusUpdate("Order Created", "OrderId123");
        producer.sendOrderStatusUpdate("Payment Received", "OrderId123");
        producer.sendOrderStatusUpdate("Shipped", "OrderId123");
        producer.sendOrderStatusUpdate("Delivered", "OrderId123");
    }
}

通過這個(gè)設(shè)置，RocketMQ 和 Spring Boot 能夠保證同一個(gè)訂單的不同狀態(tài)更新是按照發(fā)送順序被處理的。這對(duì)于需要順序一致性的業(yè)務(wù)邏輯是非常重要的。

以上就是RocketMQ保證消息的有序性的案例分享的詳細(xì)內(nèi)容，更多關(guān)于RocketMQ消息有序性的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: