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

RocketMQ 在分布式系统中的应用案例

2021-05-212.3k 阅读

RocketMQ 基础概念

RocketMQ 架构概述

RocketMQ 是一款分布式消息队列，由阿里巴巴开源，后来捐赠给 Apache 基金会成为顶级项目。它的架构主要由 NameServer、Broker、Producer 和 Consumer 组成。

  1. NameServer：是一个轻量级的服务发现与路由组件。它主要负责存储 Topic 与 Broker 的映射关系等元数据信息。NameServer 之间相互独立，没有主从关系，是一种去中心化的设计。Producer 和 Consumer 通过 NameServer 发现 Broker 的地址，进而进行消息的发送和接收。
  2. Broker：负责存储和转发消息。一个 Broker 可以存储多个 Topic 的消息，它与 NameServer 保持心跳连接，定期上报自己的存活状态和 Topic 信息。Broker 分为 Master 和 Slave 两种角色，Master 负责处理读写请求，Slave 则作为 Master 的备份，在 Master 出现故障时可以接替其工作，保证系统的高可用性。
  3. Producer：即消息生产者，负责向 Broker 发送消息。Producer 在发送消息前，会向 NameServer 获取 Topic 对应的 Broker 地址列表，然后根据负载均衡算法选择一个 Broker 进行消息发送。
  4. Consumer：即消息消费者，负责从 Broker 接收消息并进行处理。Consumer 同样需要从 NameServer 获取 Topic 对应的 Broker 地址列表，然后根据消费模式（集群消费或广播消费）从 Broker 拉取消息进行消费。

RocketMQ 消息模型

  1. Topic：主题，是消息的逻辑分类。不同的业务场景可以使用不同的 Topic 来区分消息，比如订单业务的消息可以发送到 “order - topic”，物流业务的消息可以发送到 “logistics - topic”。
  2. Queue：队列，是 Topic 的物理分区。一个 Topic 可以包含多个 Queue，这样可以提高消息的并行处理能力。例如，订单业务如果订单量很大，可以将 “order - topic” 划分为多个 Queue，不同的 Producer 可以并行地向不同的 Queue 发送订单消息，Consumer 也可以并行地从不同 Queue 拉取消息进行处理。
  3. Message：消息，是 RocketMQ 中数据的基本单元。它由消息体（Body）、消息标签（Tag）和消息键（Key）组成。消息体是实际要发送的数据内容，消息标签用于对消息进行更细粒度的分类，消息键则可以用于消息的唯一性标识和消息查询等。

RocketMQ 在分布式系统中的应用场景

异步处理

在分布式系统中，很多业务操作可能涉及到多个子系统的交互，有些操作并不需要立即得到结果。例如，用户注册成功后，系统需要发送欢迎邮件、初始化用户积分等操作。这些操作可以通过 RocketMQ 进行异步处理。

  1. 业务流程：
    • 用户在前端发起注册请求，后端注册服务处理完用户注册的核心逻辑（如将用户信息插入数据库）后，向 RocketMQ 发送一条 “用户注册成功” 的消息。
    • 邮件服务和积分服务作为消费者监听相应的 Topic。当它们收到 “用户注册成功” 的消息后，分别执行发送欢迎邮件和初始化用户积分的操作。
  2. 优点：
    • 提高系统响应速度。注册服务在发送消息后可以立即返回给用户注册成功的响应，而不需要等待邮件发送和积分初始化完成，大大缩短了用户等待时间。
    • 解耦系统模块。注册服务、邮件服务和积分服务之间通过消息队列进行通信，它们之间的耦合度降低，每个服务可以独立进行扩展和维护。

削峰填谷

在一些高并发的场景下，系统可能会面临瞬间大量的请求，如电商的秒杀活动、直播带货等。如果直接将这些请求压到后端服务，可能会导致服务过载甚至崩溃。RocketMQ 可以起到削峰填谷的作用。

  1. 业务流程：
    • 在高并发请求到达时，前端应用将请求消息发送到 RocketMQ。
    • 后端服务以固定的速率从 RocketMQ 拉取消息进行处理。例如，后端服务每秒可以处理 100 条订单消息，无论前端瞬间发送了 1000 条还是 10000 条订单消息，后端服务都按照每秒 100 条的速度从 RocketMQ 拉取消息处理，从而避免了后端服务因瞬间高并发请求而崩溃。
  2. 优点：
    • 保护后端服务。通过将高并发请求缓存在消息队列中，后端服务可以按照自身的处理能力平稳地处理请求，不会因为突发的高流量而受到冲击。
    • 提高系统稳定性。削峰填谷的机制可以让系统在高并发场景下依然保持稳定运行，保障业务的连续性。

数据分发与广播

在分布式系统中，有时需要将一条消息分发给多个不同的系统或模块进行处理。例如，电商平台的订单状态变更消息，需要通知库存系统、物流系统、财务系统等多个子系统。

  1. 业务流程：
    • 订单系统在订单状态发生变更时，向 RocketMQ 发送订单状态变更消息到特定的 Topic。
    • 库存系统、物流系统、财务系统等作为消费者订阅该 Topic。当它们收到订单状态变更消息后，各自根据自身业务逻辑进行相应的处理，如库存系统更新库存、物流系统更新物流状态、财务系统更新订单财务信息等。
  2. 优点：
    • 实现数据的高效分发。通过消息队列的广播机制，一条消息可以同时被多个消费者获取并处理，减少了系统之间的数据同步成本。
    • 易于扩展。如果后续有新的子系统需要关注订单状态变更，只需要订阅相应的 Topic 即可，不需要对订单系统和其他现有子系统进行大量的代码修改。

RocketMQ 在分布式电商系统中的应用案例

订单系统中的异步处理

  1. 业务场景：
    • 在电商系统中，用户下单后，除了要完成订单的基本创建操作（如记录订单信息到数据库），还需要进行库存扣减、订单消息通知（如短信通知用户订单已生成）等操作。其中库存扣减和订单消息通知并不需要与订单创建操作同步完成，可以通过异步方式处理。
  2. 系统架构：
    • 订单服务：负责接收用户下单请求，处理订单创建的核心逻辑，然后向 RocketMQ 发送 “订单创建成功” 的消息。
    • 库存服务：作为消费者监听 RocketMQ 中 “订单创建成功” 相关 Topic，收到消息后进行库存扣减操作。
    • 消息通知服务：同样作为消费者监听 “订单创建成功” 相关 Topic，收到消息后向用户发送订单已生成的短信通知。
  3. 代码示例：
    • 订单服务发送消息（Java 示例）：
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", "订单创建成功消息".getBytes("UTF - 8"));
        // 发送消息
        SendResult sendResult = producer.send(message);
        System.out.println("发送结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}
  • 库存服务消费消息（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 StockConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        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()));
                    // 执行库存扣减逻辑
                    // 这里简单模拟库存扣减，实际业务中需要与库存数据库交互
                    System.out.println("执行库存扣减操作");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("库存消费者已启动");
    }
}
  • 消息通知服务消费消息（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 NotificationConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("notification - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        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()));
                    // 执行短信通知逻辑
                    // 这里简单模拟短信通知，实际业务中需要调用短信发送接口
                    System.out.println("向用户发送订单已生成短信通知");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("消息通知消费者已启动");
    }
}

电商促销活动中的削峰填谷

  1. 业务场景：
    • 在电商促销活动期间，如 “双 11”、“618” 等，大量用户同时涌入平台下单，瞬间会产生海量的订单请求。如果直接将这些请求压到订单处理服务，可能会导致服务因过载而崩溃。
  2. 系统架构：
    • 前端应用：在促销活动期间，用户下单请求到达前端应用后，前端应用将订单请求消息发送到 RocketMQ。
    • 订单处理服务：以固定的速率从 RocketMQ 拉取订单消息进行处理。订单处理服务可以根据自身的处理能力，设置合适的拉取速率，如每秒拉取 100 条订单消息进行处理。
  3. 代码示例：
    • 前端应用发送订单消息（Java 示例，简化模拟）：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

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

        // 模拟大量订单请求，这里简单循环发送 1000 条消息
        for (int i = 0; i < 1000; i++) {
            String orderMessage = "促销活动订单请求 " + i;
            Message message = new Message("promotion - order - topic", "order - tag", orderMessage.getBytes("UTF - 8"));
            SendResult sendResult = producer.send(message);
            System.out.println("发送订单消息 " + i + " 结果：" + sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}
  • 订单处理服务消费订单消息（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;
import java.util.concurrent.TimeUnit;

public class PromotionOrderProcessor {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("promotion - order - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        consumer.subscribe("promotion - 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()));
                    // 模拟订单处理逻辑，这里简单睡眠 10 毫秒
                    try {
                        TimeUnit.MILLISECONDS.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("促销活动订单处理消费者已启动");
    }
}

订单状态变更的数据分发

  1. 业务场景：
    • 在电商系统中，订单状态发生变更（如已付款、已发货、已完成等）时，需要通知多个相关子系统，如库存系统、物流系统、财务系统等，以便它们进行相应的业务处理。
  2. 系统架构：
    • 订单服务：当订单状态发生变更时，向 RocketMQ 发送订单状态变更消息到 “order - status - change - topic”。
    • 库存系统：作为消费者订阅 “order - status - change - topic”，当收到订单已付款的消息时，可能需要解锁库存；收到订单已完成的消息时，可能需要更新库存的销售记录等。
    • 物流系统：订阅 “order - status - change - topic”，收到订单已发货的消息后，更新物流状态为 “已发货” 等。
    • 财务系统：订阅 “order - status - change - topic”，收到订单已完成的消息后，进行财务结算等操作。
  3. 代码示例：
    • 订单服务发送订单状态变更消息（Java 示例）：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

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

        // 模拟订单状态变更为已付款
        String orderStatusMessage = "订单已付款";
        Message message = new Message("order - status - change - topic", "payment - tag", orderStatusMessage.getBytes("UTF - 8"));
        SendResult sendResult = producer.send(message);
        System.out.println("发送订单状态变更消息结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}
  • 库存系统消费订单状态变更消息（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 StockSystemConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock - system - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        consumer.subscribe("order - status - change - topic", "payment - 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()));
                    // 执行库存解锁逻辑
                    System.out.println("执行库存解锁操作");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("库存系统消费者已启动");
    }
}
  • 物流系统消费订单状态变更消息（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 LogisticsSystemConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("logistics - system - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        consumer.subscribe("order - status - change - topic", "shipped - 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()));
                    // 执行更新物流状态逻辑
                    System.out.println("更新物流状态为已发货");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("物流系统消费者已启动");
    }
}
  • 财务系统消费订单状态变更消息（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 FinanceSystemConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("finance - system - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        consumer.subscribe("order - status - change - topic", "completed - 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()));
                    // 执行财务结算逻辑
                    System.out.println("执行财务结算操作");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("财务系统消费者已启动");
    }
}

RocketMQ 在分布式系统中的部署与优化

单机房部署

  1. 部署架构：
    • 在单机房部署场景下，NameServer 可以部署多个实例以提高可用性，它们之间相互独立，没有主从关系。Broker 同样可以部署多个实例，分为 Master 和 Slave，Master 负责处理读写请求，Slave 作为备份。Producer 和 Consumer 与 NameServer 和 Broker 位于同一机房内。
  2. 部署步骤：
    • 安装 NameServer：下载 RocketMQ 安装包，解压后进入 bin 目录，执行 mqnamesrv 命令启动 NameServer。可以通过修改 conf/broker.conf 文件中的 namesrvAddr 配置项来指定 NameServer 地址。
    • 安装 Broker：同样在 RocketMQ 安装包解压目录下，修改 conf/broker.conf 文件，配置 Broker 的相关信息，如 brokerName、brokerId、namesrvAddr 等。然后在 bin 目录下执行 mqbroker -c ../conf/broker.conf 命令启动 Broker。
    • 配置 Producer 和 Consumer：在 Producer 和 Consumer 的代码中，通过 producer.setNamesrvAddr("nameserver1:9876;nameserver2:9876") 和 consumer.setNamesrvAddr("nameserver1:9876;nameserver2:9876") 来设置 NameServer 地址。

多机房部署

  1. 部署架构：
    • 在多机房部署时，每个机房都有自己独立的 NameServer 集群和 Broker 集群。Producer 和 Consumer 可以根据就近原则选择机房内的 NameServer 和 Broker 进行消息的发送和接收。不同机房之间的 Broker 可以通过同步或异步的方式进行数据复制，以保证数据的一致性。
  2. 部署要点：
    • 跨机房网络配置：需要确保不同机房之间有可靠的网络连接，以保证 Broker 之间的数据复制和同步能够正常进行。
    • NameServer 配置：每个机房的 NameServer 集群需要配置相互之间的地址，以便 Producer 和 Consumer 在不同机房切换时能够获取到正确的 Broker 地址。
    • Broker 数据同步：可以采用同步复制或异步复制的方式。同步复制保证了数据的强一致性，但会降低系统的性能；异步复制则可以提高系统的性能，但可能会存在数据丢失的风险，需要根据业务需求进行选择。

RocketMQ 性能优化

  1. Producer 优化：
    • 批量发送消息：Producer 可以将多条消息批量发送到 Broker，减少网络 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++) {
            String orderMessage = "批量订单消息 " + i;
            Message message = new Message("batch - order - topic", "order - tag", orderMessage.getBytes("UTF - 8"));
            messages.add(message);
        }
        SendResult sendResult = producer.send(messages);
        System.out.println("批量发送订单消息结果：" + sendResult);

        // 关闭生产者
        producer.shutdown();
    }
}
  • 选择合适的发送方式：Producer 有同步发送、异步发送和单向发送三种方式。同步发送可靠性高，但会阻塞线程；异步发送可以提高系统性能，但需要处理回调逻辑；单向发送速度最快，但不保证消息发送成功，适用于对消息可靠性要求不高的场景。可以根据业务需求选择合适的发送方式。
  1. Consumer 优化：
    • 并发消费：Consumer 可以设置并发消费线程数，提高消息的消费速度。例如，在订单处理服务中，可以设置多个线程同时处理订单消息。
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 ConcurrentOrderConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("concurrent - order - consumer - group");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        consumer.subscribe("batch - order - topic", "order - tag");

        // 设置并发消费线程数
        consumer.setConsumeThreadMin(10);
        consumer.setConsumeThreadMax(20);

        // 注册消息监听器
        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("并发订单消费者已启动");
    }
}
  • 合理设置消费模式：根据业务需求选择集群消费或广播消费模式。集群消费模式下，同一个 Consumer Group 中的多个 Consumer 实例共同消费消息，适用于对消息处理有负载均衡需求的场景；广播消费模式下，同一个 Consumer Group 中的每个 Consumer 实例都会消费到所有消息，适用于需要所有实例都处理消息的场景，如系统配置更新通知等。
  1. Broker 优化：
    • 存储优化：可以调整 Broker 的存储配置，如选择合适的存储介质（SSD 比机械硬盘读写速度快），优化存储目录结构等，提高消息的存储和读取性能。
    • 网络优化：合理配置 Broker 的网络参数，如调整 TCP 缓冲区大小、优化网络拓扑等，减少网络延迟，提高消息传输效率。

RocketMQ 在分布式系统中的常见问题及解决方法

消息丢失问题

  1. 原因分析：
    • Producer 发送消息失败：可能由于网络故障、Broker 不可用等原因导致 Producer 发送消息失败，而应用程序没有正确处理发送失败的情况，从而导致消息丢失。
    • Broker 数据未持久化：如果 Broker 配置了异步刷盘或未开启持久化功能，在 Broker 宕机时，内存中的消息可能会丢失。
    • Consumer 消费失败：Consumer 在消费消息过程中出现异常，如代码逻辑错误、系统资源不足等，导致消息未能成功处理，但又向 Broker 发送了消费成功的确认，从而导致消息丢失。
  2. 解决方法：
    • Producer 端：在发送消息时，正确处理发送失败的情况，如进行重试。可以设置 producer.setRetryTimesWhenSendFailed(3) 来设置发送失败时的重试次数。
    • Broker 端：开启同步刷盘或使用高可靠的持久化策略。在 conf/broker.conf 文件中，设置 flushDiskType = SYNC_FLUSH 来开启同步刷盘，确保消息在写入内存的同时也写入磁盘，提高消息的可靠性。
    • Consumer 端：在消费消息时，进行异常捕获和处理。如果消费失败，不要立即向 Broker 发送消费成功的确认，而是根据业务需求进行重试或记录到重试队列中，后续进行人工处理。

消息重复问题

  1. 原因分析：
    • 网络抖动：在消息发送和接收过程中，由于网络抖动，可能导致消息重复发送或重复接收。例如，Producer 发送消息后，在等待 Broker 确认时网络出现短暂中断，Producer 未收到确认而进行重试，而 Broker 实际上已经成功接收了第一次发送的消息，从而导致消息重复。
    • Consumer 消费确认问题：Consumer 消费消息后，向 Broker 发送消费确认时出现网络问题，Broker 未收到确认，而 Consumer 认为已经确认成功，此时 Broker 可能会再次向 Consumer 发送该消息，导致消息重复消费。
  2. 解决方法：
    • 幂等性处理：在 Consumer 端实现幂等性消费，即对于相同的消息，无论消费多少次，其结果都是一致的。例如，在订单处理服务中，对于 “创建订单” 的消息，在消费时可以先根据订单号查询订单是否已经创建，如果已经创建则不再重复创建。
    • 消息去重：可以在消息中添加唯一标识（如 UUID），Consumer 在消费消息时，先根据唯一标识判断该消息是否已经消费过，如果已经消费过则直接丢弃。可以使用 Redis 等缓存来记录已经消费过的消息唯一标识。

消息堆积问题

  1. 原因分析：
    • Consumer 消费能力不足：Consumer 的处理速度跟不上 Producer 发送消息的速度，导致消息在 Broker 中不断堆积。例如，在电商促销活动期间，订单消息大量涌入，而订单处理服务由于系统资源限制等原因，无法及时处理所有订单消息。
    • Broker 存储容量限制：如果 Broker 的存储容量有限，当消息堆积过多，超过 Broker 的存储容量时，可能会导致新的消息无法写入。
  2. 解决方法：
    • 提高 Consumer 消费能力：可以通过增加 Consumer 实例数量、提高 Consumer 的并发处理能力（如增加并发线程数）等方式来提高 Consumer 的消费速度。
    • 优化存储：如果是 Broker 存储容量问题，可以考虑扩展 Broker 的存储容量，如增加磁盘空间或采用分布式存储方案。同时，也可以定期清理历史消息，释放存储空间。

通过以上对 RocketMQ 在分布式系统中的应用案例、部署优化以及常见问题解决方法的介绍，希望能帮助开发者更好地理解和应用 RocketMQ，构建更加稳定、高效的分布式系统。