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

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 计算队列 ID
        return Math.abs(msg.getKey().hashCode()) % this.queueSize;
    }
}

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

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

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

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保证同一个订单的更新在同一个队列
        SendResult result = rocketMQTemplate.syncSendOrderly("order-topic", orderStatus, orderId);
        System.out.println("Message sent, result: " + result.getSendStatus());
    }
}

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) {
        // 实现订单更新处理逻辑
        System.out.println("Processing order status update: " + status);
    }
}

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");
    }
}