MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

RocketMQ 在电商系统中的应用实践

2023-07-105.2k 阅读

电商系统中的消息队列需求分析

在电商系统中，存在众多异步处理场景，这些场景对系统的性能、可靠性和扩展性提出了很高的要求。消息队列作为一种可靠的异步通信机制，在电商系统中扮演着至关重要的角色。

高并发下单场景

电商大促活动时，瞬间会有海量的下单请求涌入。如果所有请求都同步处理，比如库存扣减、订单生成、支付处理等操作都在一个事务中同步执行，数据库很可能会因为高并发而不堪重负，导致响应时间变长甚至系统崩溃。此时，引入消息队列可以将下单请求先放入队列中，系统从队列中逐步取出请求进行处理，实现削峰填谷，减轻数据库等核心系统的压力。

异步业务处理

电商系统中有许多业务操作不需要实时返回结果，如订单成功后的积分发放、物流信息推送、优惠券发放等。这些操作如果与核心业务流程同步执行，会延长用户等待时间，降低用户体验。通过消息队列，这些异步任务可以被发送到队列中，由专门的消费者进行处理，主业务流程能够快速返回响应，提高系统整体的响应性能。

系统解耦

电商系统是一个复杂的生态，包含多个子系统，如订单系统、库存系统、支付系统、物流系统等。不同子系统之间存在大量的交互。例如，当订单创建成功后，需要通知库存系统扣减库存，通知物流系统准备发货等。如果这些子系统之间直接进行调用，会导致系统耦合度极高，一个子系统的变更可能会影响到其他多个子系统。使用消息队列可以实现子系统之间的解耦，各个子系统只需要关注消息的发送和接收，而不需要关心其他子系统的具体实现和状态，提高系统的可维护性和扩展性。

RocketMQ 简介

RocketMQ 是一款由阿里巴巴开源的分布式消息中间件，具有高性能、高可靠、高扩展性等特点，非常适合在电商这样的复杂业务场景中使用。

RocketMQ 的架构

RocketMQ 主要由 NameServer、Broker、Producer 和 Consumer 组成。

  • NameServer：是一个轻量级的元数据服务器，主要负责 Broker 的注册与发现。每个 NameServer 之间相互独立，不进行数据同步。Producer 和 Consumer 通过 NameServer 发现 Broker 的地址信息。
  • Broker：负责消息的存储、转发等核心功能。Broker 可以分为 Master 和 Slave 两种角色，Master 负责处理读写请求，Slave 则作为 Master 的备份，在 Master 出现故障时可以切换为 Master 继续提供服务，保证系统的高可用性。
  • Producer：即消息生产者，负责将业务系统中的消息发送到 RocketMQ 的 Broker 中。Producer 可以分为同步发送、异步发送和单向发送等不同的发送模式，以满足不同业务场景的需求。
  • Consumer：即消息消费者，负责从 Broker 中拉取消息并进行处理。Consumer 支持集群消费和广播消费两种模式，集群消费模式下，多个 Consumer 实例共同消费一组消息，每条消息只会被其中一个 Consumer 实例处理；广播消费模式下，每条消息会被集群中的所有 Consumer 实例处理。

RocketMQ 的核心特性

  • 高吞吐量：RocketMQ 采用了多种优化技术，如零拷贝、异步刷盘等，能够实现极高的消息吞吐量，满足电商系统高并发的消息处理需求。
  • 高可靠性：通过 Master - Slave 架构以及数据多副本机制，保证了消息的可靠存储和传输。即使部分 Broker 节点出现故障，也不会导致消息丢失，确保了电商业务数据的完整性。
  • 分布式扩展性：RocketMQ 支持水平扩展，可以通过增加 Broker 节点来提高系统的整体处理能力，适应电商系统业务量不断增长的需求。

RocketMQ 在电商下单流程中的应用

下单消息的发送

在电商下单流程中，当用户提交订单后，系统首先将下单消息发送到 RocketMQ 中。以下是使用 Java 语言结合 RocketMQ 客户端发送下单消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class OrderProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("order_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息
        Message message = new Message("order_topic", "order_tag", "order_key", "下单消息内容".getBytes("UTF-8"));
        // 发送消息
        SendResult sendResult = producer.send(message);
        System.out.println("发送结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}

在上述代码中，首先创建了一个 DefaultMQProducer 实例，并指定了生产者组名称 order_producer_group。然后设置了 NameServer 的地址为 localhost:9876，启动生产者。接着创建了一条消息，指定了消息所属的主题 order_topic、标签 order_tag 和消息键 order_key，并设置了消息内容。最后使用 producer.send 方法发送消息，并输出发送结果。发送完成后，关闭生产者。

订单消息的接收与处理

在 RocketMQ 中，订单消息发送到指定的主题和队列后，需要有消费者来接收并处理这些消息。以下是使用 Java 语言实现订单消息消费者的代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class OrderConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅主题和标签
        consumer.subscribe("order_topic", "order_tag");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("收到消息：" + new String(msg.getBody()));
                    // 处理订单消息，例如扣减库存、生成订单记录等业务逻辑
                    // 这里省略具体业务逻辑代码
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("订单消费者启动成功");
    }
}

在这段代码中，首先创建了 DefaultMQPushConsumer 实例，并指定了消费者组名称 order_consumer_group。设置 NameServer 地址后，通过 subscribe 方法订阅了 order_topic 主题下的 order_tag 标签的消息。接着注册了一个 MessageListenerConcurrently 消息监听器，在监听器的 consumeMessage 方法中，遍历接收到的消息列表，输出消息内容，并在此处可以编写具体的订单处理业务逻辑，如扣减库存、生成订单记录等。处理完成后，返回 ConsumeConcurrentlyStatus.CONSUME_SUCCESS 表示消息处理成功。最后启动消费者。

RocketMQ 在库存管理中的应用

库存扣减消息的发送

当订单生成后，需要通知库存系统扣减相应的商品库存。这可以通过发送库存扣减消息到 RocketMQ 来实现。以下是发送库存扣减消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class StockProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("stock_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息
        Message message = new Message("stock_topic", "stock_tag", "stock_key", "库存扣减消息内容".getBytes("UTF-8"));
        // 发送消息
        SendResult sendResult = producer.send(message);
        System.out.println("发送结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}

与下单消息发送类似，这里创建了 stock_producer_group 生产者组的 DefaultMQProducer 实例，设置 NameServer 地址后启动生产者。创建消息时指定了 stock_topic 主题、stock_tag 标签和 stock_key 消息键，并设置消息内容为库存扣减相关信息，最后发送消息并关闭生产者。

库存扣减消息的接收与处理

库存系统作为消费者，从 RocketMQ 中接收库存扣减消息并进行处理。以下是库存扣减消息消费者的代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class StockConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock_consumer_group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅主题和标签
        consumer.subscribe("stock_topic", "stock_tag");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("收到库存扣减消息：" + new String(msg.getBody()));
                    // 执行库存扣减业务逻辑，例如更新数据库库存字段等
                    // 这里省略具体业务逻辑代码
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("库存消费者启动成功");
    }
}

此代码创建了 stock_consumer_group 消费者组的 DefaultMQPushConsumer 实例，设置 NameServer 地址并订阅 stock_topic 主题下的 stock_tag 标签消息。在消息监听器的 consumeMessage 方法中，接收并输出库存扣减消息，然后执行实际的库存扣减业务逻辑，如更新数据库中的库存字段等，处理完成后返回消费成功状态并启动消费者。

RocketMQ 在物流信息推送中的应用

物流信息推送消息的发送

当订单状态更新为已发货时，需要向用户推送物流信息。可以通过发送物流信息推送消息到 RocketMQ 来触发推送操作。以下是发送物流信息推送消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class LogisticsProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("logistics_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息
        Message message = new Message("logistics_topic", "logistics_tag", "logistics_key", "物流信息推送消息内容".getBytes("UTF-8"));
        // 发送消息
        SendResult sendResult = producer.send(message);
        System.out.println("发送结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}

这里创建了 logistics_producer_group 生产者组的 DefaultMQProducer 实例，设置 NameServer 地址后启动生产者。创建消息时指定了 logistics_topic 主题、logistics_tag 标签和 logistics_key 消息键，并设置消息内容为物流信息推送相关信息，最后发送消息并关闭生产者。

物流信息推送消息的接收与处理

物流信息推送服务作为消费者，从 RocketMQ 中接收物流信息推送消息并进行实际的推送操作。以下是物流信息推送消息消费者的代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class LogisticsConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("logistics_consumer_group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅主题和标签
        consumer.subscribe("logistics_topic", "logistics_tag");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("收到物流信息推送消息：" + new String(msg.getBody()));
                    // 执行物流信息推送业务逻辑，例如调用短信或推送平台接口等
                    // 这里省略具体业务逻辑代码
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("物流信息推送消费者启动成功");
    }
}

此代码创建了 logistics_consumer_group 消费者组的 DefaultMQPushConsumer 实例，设置 NameServer 地址并订阅 logistics_topic 主题下的 logistics_tag 标签消息。在消息监听器的 consumeMessage 方法中，接收并输出物流信息推送消息，然后执行实际的物流信息推送业务逻辑，如调用短信平台或推送平台接口向用户发送物流信息等，处理完成后返回消费成功状态并启动消费者。

RocketMQ 在电商系统中的高级应用与优化

事务消息的应用

在电商系统中，有些业务场景需要保证多个操作的原子性，例如订单创建成功后，既要扣减库存，又要进行支付操作，这两个操作必须要么都成功，要么都失败。RocketMQ 提供的事务消息可以满足这种需求。

事务消息的发送过程分为两个阶段：

  1. Half 消息阶段：Producer 先发送一条 Half 消息到 Broker，此时这条消息对 Consumer 不可见。
  2. 事务提交阶段：Producer 执行本地事务，根据本地事务的执行结果向 Broker 发送 Commit 或 Rollback 指令。如果发送 Commit 指令，Broker 将 Half 消息标记为可消费状态，Consumer 可以接收并处理该消息；如果发送 Rollback 指令，Broker 将删除 Half 消息。

以下是使用 RocketMQ 事务消息实现订单创建与库存扣减原子性操作的代码示例：

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.*;

public class TransactionOrderProducer {
    public static void main(String[] args) throws Exception {
        // 创建事务生产者实例
        TransactionMQProducer producer = new TransactionMQProducer("transaction_order_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");

        // 创建线程池
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000));
        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 执行本地事务，例如创建订单和扣减库存
                // 这里省略具体业务逻辑代码
                // 如果本地事务执行成功，返回 LocalTransactionState.COMMIT_MESSAGE
                // 如果本地事务执行失败，返回 LocalTransactionState.ROLLBACK_MESSAGE
                // 如果无法确定本地事务执行结果，返回 LocalTransactionState.UNKNOW
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 检查本地事务状态，用于 Broker 回查
                // 这里省略具体业务逻辑代码
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        // 设置线程池
        producer.setExecutorService(executorService);
        // 启动生产者
        producer.start();

        // 创建消息
        Message message = new Message("transaction_order_topic", "transaction_order_tag", "transaction_order_key", "事务订单消息内容".getBytes("UTF-8"));
        // 发送事务消息
        producer.sendMessageInTransaction(message, null);

        // 线程睡眠，确保消息发送完成
        Thread.sleep(10000);
        // 关闭生产者
        producer.shutdown();
        // 关闭线程池
        executorService.shutdown();
    }
}

在上述代码中，创建了 TransactionMQProducer 实例，并设置了生产者组名称和 NameServer 地址。创建了一个线程池用于执行事务操作。然后设置了 TransactionListener 事务监听器，在 executeLocalTransaction 方法中执行本地事务，如创建订单和扣减库存等业务逻辑，并根据执行结果返回相应的 LocalTransactionState 状态。checkLocalTransaction 方法用于 Broker 回查本地事务状态。启动生产者后，发送事务消息，并在最后关闭生产者和线程池。

消息可靠性保障

在电商系统中，消息的可靠性至关重要，任何消息的丢失都可能导致业务异常。RocketMQ 通过多种机制来保障消息的可靠性。

  • 消息持久化：RocketMQ 的 Broker 默认采用异步刷盘的方式将消息持久化到磁盘中，保证即使 Broker 重启，消息也不会丢失。也可以通过配置改为同步刷盘，进一步提高消息持久化的可靠性，但会牺牲一定的性能。
  • Master - Slave 架构：Broker 采用 Master - Slave 架构，Master 节点负责处理读写请求，Slave 节点作为备份。当 Master 节点出现故障时，Slave 节点可以切换为 Master 继续提供服务，保证消息的可用性和可靠性。
  • 消息重试：当 Consumer 消费消息失败时，RocketMQ 会自动进行消息重试。默认情况下，Consumer 会进行 16 次重试，每次重试的间隔时间会逐渐延长。如果 16 次重试后仍然失败，消息会被发送到死信队列，供后续人工处理。

性能优化

为了满足电商系统高并发的性能需求，对 RocketMQ 进行性能优化是必要的。

  • 批量发送消息：Producer 可以采用批量发送消息的方式，减少网络 I/O 开销，提高消息发送性能。例如：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.util.ArrayList;
import java.util.List;

public class BatchOrderProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("batch_order_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息列表
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("batch_order_topic", "batch_order_tag", ("订单消息" + i).getBytes("UTF-8"));
            messages.add(message);
        }

        // 批量发送消息
        SendResult sendResult = producer.send(messages);
        System.out.println("批量发送结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}

在上述代码中，创建了一个消息列表，将多个消息添加到列表中，然后使用 producer.send 方法批量发送这些消息。

  • 异步发送消息：对于一些对实时性要求不高的业务场景，Producer 可以采用异步发送消息的方式，提高系统的并发处理能力。例如：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class AsyncOrderProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("async_order_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息
        Message message = new Message("async_order_topic", "async_order_tag", "async_order_key", "异步订单消息内容".getBytes("UTF-8"));
        // 异步发送消息
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步发送成功：" + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("异步发送失败：" + e);
            }
        });

        // 线程睡眠，确保消息发送完成
        Thread.sleep(1000);
        // 关闭生产者
        producer.shutdown();
    }
}

在这段代码中，通过 producer.send 方法的第二个参数传入一个 SendCallback 回调对象，在消息发送成功或失败时会分别调用 onSuccess 和 onException 方法进行相应的处理。采用异步发送方式，Producer 不会阻塞等待消息发送结果，提高了系统的并发性能。

通过以上在电商系统不同业务场景中的应用以及高级应用与优化，RocketMQ 能够有效地满足电商系统对消息队列的高性能、高可靠和高扩展性的需求，为电商业务的稳定运行和发展提供有力支持。在实际应用中，需要根据电商系统的具体业务特点和需求，合理配置和使用 RocketMQ，以达到最佳的性能和业务效果。同时，随着电商业务的不断发展和变化，也需要持续关注 RocketMQ 的更新和优化，不断完善系统的消息处理机制。