SpringCloud Alibaba】（十三）学习 RocketMQ 消息队列

1、MQ 使用场景与选型对比

1.1、MQ 的使用场景

MQ 的英文全称是 Message Queue，翻译成中文就是消息队列，队列实现了先进先出（FIFO）的消息模型。通过消息队列，我们可以实现多个进程之间的通信，例如，可以实现多个微服务之间的消息通信。MQ 的最简模型就是生产者生产消息，将消息发送到 MQ，消息消费者订阅 MQ，消费消息

MQ的使用场景通常包含：异步解耦、流量削峰

1.2、引入 MQ 后的注意事项

引入MQ最大的优点就是异步解耦和流量削峰，但是引入 MQ 后也有很多需要注意的事项和问题，主要包括：系统的整体可用性降低、系统的复杂度变高、引入了消息一致性的问题

• 系统的整体可用性降低：在对一个系统进行架构设计时，引入的外部依赖越多，系统的稳定性和可用性就会降低。系统中引入了MQ，部分业务就会出现强依赖MQ的现象，此时，如果MQ宕机，则部分业务就会变得不可用。所以，引入MQ时，我们就要考虑如何实现MQ的高可用。
• 系统的复杂度变高：引入 MQ 后，会使之前的同步接口调用变成通过 MQ 的异步调用，在实际的开发过程中，异步调用会比同步调用复杂的多。并且异步调用出现问题后，重现问题，定位问题和解决问题都会比同步调用复杂的多。并且引入 MQ 后，还要考虑如何保证消息的顺序等问题
• 消息一致性问题 ：引入 MQ 后，不得不考虑的一个问题就是消息的一致性问题。这期间就要考虑如何保证消息不丢失，消息幂等和消息数据处理的幂等性问题

1.3、MQ 选型对比

目前，在行业内使用的比较多的 MQ 包含 RabbitMQ、Kafka 和 RocketMQ。这里，我将三者的对比简单整理了个表格，如下所示：

2、下载、安装 RocketMQ 及 RocketMQ 控制台

2.1、下载安装 RocketMQ

Apache RocketMQ开发者指南

Windows 部署 RocketMQ

RocketMQ 下载、安装过程：

1. 下载、解压。下载地址：二进制版本 4.9.2 官方下载

修改配置文件 conf/broker.conf：

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH

# 自动创建Topic
autoCreateTopicEnable=true
# nameServ地址
namesrvAddr=127.0.0.1:9876
# 存储路径
storePathRootDir=D:/dev/rocketmq-4.9.2/data/dataDir
# commitLog路径
storePathCommitLog=D:/dev/rocketmq-4.9.2/data/dataDir/commitlog
# 消息队列存储路径
storePathConsumeQueue=D:/dev/rocketmq-4.9.2/data/dataDir/consumequeue
# 消息索引存储路径
storePathIndex=D:/dev/rocketmq-4.9.2/data/dataDir/index
# checkpoint文件路径
storeCheckpoint=D:/dev/rocketmq-4.9.2/data/dataDir/checkpoint
# abort文件存储路径
abortFile=D:/dev/rocketmq-4.9.2/data/dataDir/abort

2. 配置 ROCKETMQ 环境变量：否则，启动报错

   • 先添加一个环境变量 ROCKETMQ_HOME
     
   • 在 Path 中进行添加

3. 内存分配设置【可选】

①：编辑 server 启动文件：bin/runserver.cmd

set "JAVA_OPT=%JAVA_OPT% -server -Xms256m -Xmx512m -Xmn512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

②：编辑 broker 启动文件：bin/runbroker.cmd

set "JAVA_OPT=%JAVA_OPT% -server -Xms256m -Xmx256m -Xmn128m"

4. 修改日志存储默认路径：conf/logback_broker.xml、conf/logback_namesrv.xml、conf/logback_tools.xml

5. 启动 NameServer

命令行执行：

mqnamesrv.cmd

打印出如下信息，说明 RocketMQ 的 NameServer 启动成功了：

6. 启动 Broker

命令行执行：

mqbroker.cmd -n localhost:9876

打印出如下信息，说明 RocketMQ 的 Broker 服务启动成功了：

2.2、测试 RocketMQ 环境

RocketMQ 内置了大量的测试案例，并且这些测试案例可以通过 RocketMQ 的 bin 目录下的 tools.cmd 命令进行测试

1、启动生产者程序向 RocketMQ 发送消息

重新打开 cmd 命令行，进入 RocketMQ 的 bin 目录，在命令行输入如下命令调用 RocketMQ 自带的生产者程序向 RocketMQ 发送消息：

set NAMESRV_ADDR=localhost:9876
tools.cmd org.apache.rocketmq.example.quickstart.Producer

可以看到，执行完上述两条命令后，生产者程序开始向 RocketMQ 发送消息：

2、启动消费者程序消费 RocketMQ 中的消息

重新打开 cmd 命令行，进入 RocketMQ 的 bin 目录，在命令行输入如下命令调用 RocketMQ 自带的消费者程序消费 RocketMQ 中的消息

set NAMESRV_ADDR=localhost:9876
tools.cmd org.apache.rocketmq.example.quickstart.Consumer

可以看到，执行完上述两条命令后，消费者程序开始消费 RocketMQ 中的消息：

2.3、RocketMQ 控制台【图形化管理控制台】

2.3.1、下载、安装

RocketMQ 控制台本质上是一个 SpringBoot 程序，启动后默认监听的端口是 8080。RocketMQ 的新版控制台已经从 RocketMQ 的 rocketmq-externals 项目中分离出来了。也就是说，新版的 RocketMQ 控制台已经从 https://github.com/apache/rocketmq-externals 链接所示的项目中分离出来，新版控制台的链接地址为：https://github.com/apache/rocketmq-dashboard

1、下载 RocketMQ 控制台源码
2、修改配置：src/main/resources/application.properties

• 端口：7000
• namesrvAddr：localhost:9876
• dataPath：D:/dev/rocket-mq-master/data

3、打开 cmd 命令行，进入 RocketMQ 控制台源码的根目录，输入如下 Maven 命令开始编 RocketMQ 控制台的源码：

mvn clean install -Dmaven.test.skip=true

4、编译完成后，会在 RocketMQ 控制台源码的根目录下生成 target 目录，进入 target 目录，可以看到生成了 rocketmq-dashboard-1.0.1-SNAPSHOT.jar 文件，如下所示

5、重新打开 cmd 命令行，进入 rocketmq-dashboard-1.0.0.jar 文件所在的目录，在命令行直接输入如下命令启动 RocketMQ 控制台程序

java -jar rocketmq-console-ng-1.0.0.jar

2.3.2、验证 RocketMQ 控制台

在浏览器中输入 http://localhost:7000 后，出现如下画面说明 RocketMQ 启动成功【可切换语言】：

选择【Topic】菜单想后可以看到目前 RocketMQ 中存在一个名称为 TopicTest 的主题：

点击 TopicTest 主题的状态按钮，如下所示：

可以看到，正确显示出了 TopicTest 主题的消息队列信息，说明 RocketMQ 控制台启动成功了

3、RocketMQ 快速入门

3.1、导入 RocketMQ 依赖

在用户微服务 shop-user 的 pom.xml 中，添加 RocketMQ 相关的依赖，如下所示：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.3</version>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.5.2</version>
</dependency>

3.2、编写生产者代码

在用户微服务的 src/test/java 目录下新建 com.zzc.rocketmq.test包，在包下创建 RocketMQProducer 类，作为 RocketMQ 的生产者，代码如下所示：

public class RocketMQProducer {
    public static void main(String[] args) throws Exception {
        // 1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
        // 2.指定Nameserver地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 3.启动producer
        producer.start();
        // 4.构建消息对象
        Message message = new Message("bingheTopic", "bingheTag", "Hello RocketMQ".getBytes());
        System.out.println("生产者发出的消息为：" + JSONObject.toJSONString(message));
        // 5.发送消息并接收结果
        SendResult sendResult = producer.send(message);
        System.out.println("生产者收到的发送结果信息为：" + JSONObject.toJSONString(sendResult));
        // 6.关闭生产者
        producer.shutdown();
    }
}

3.3、编写消费者代码

在 com.zzc.rocketmq.test 包下新建 RocketMQConsumer 类，作为 RocketMQ 的消费者，代码如下所示：

public class RocketMQConsumer {
    public static void main(String[] args) throws Exception {
        // 1.创建消息消费者 consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
        // 2.指定Nameserver地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 3.订阅 testTopic主题
        consumer.subscribe("testTopic", "*");
        // 4.设置消息监听，当收到消息时 RocketMQ 会回调消息监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 打印消息消费者收到的RocketMQ消息
                System.out.println("消费者收到的消息为：" + list);
                // 返回消息消费成功的标识
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 5.启动消费者
        System.out.println("消费者启动成功");
        consumer.start();
    }
}

3.4、测试消息的生产与消费

1、为了便于观察，这里我们先启动消费者程序 RocketMQConsumer，启动 RocketMQConsumer后会在 IDEA 的控制台打印如下信息：

2、运行生产者程序 RocketMQProducer，运行后 RocketMQProducer 程序控制台会输出如下信息：

说明生产者程序 RocketMQProducer 成功将消息发送到 RocketMQ

3、接下来，再看下消费者程序 RocketMQConsumer 的控制台，如下所示：

说明生产者发送到 RocketMQ 的消息，被消费者成功消费到了

4、集成 RocketMQ

在项目中模拟一个用户成功下单后，为用户发送通知，通知用户下单成功的逻辑。

具体的流程：

下单成功后将订单的信息发送到 RocketMQ，然后用户微服务订阅 RocketMQ 的消息，接收到消息后进行打印

• 用户微服务：消费者
• 订单微服务：生产者

4.1、用户微服务集成 RocketMQ

1、在用户微服务 shop-user 导入了 RocketMQ 的依赖

2、在用户微服务 shop-user 的 application.yml 文件中添加如下 RocketMQ 的配置

rocketmq:
  name-server: 127.0.0.1:9876

3、在用户微服务 shop-user 中创建 com.zzc.user.mq 包，在包下创建 RocketConsumeListener，实现 org.apache.rocketmq.spring.core.RocketMQListener 接口，具体代码如下所示：

@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user-group", topic = "order-topic")
public class RocketConsumeListener implements RocketMQListener<Order> {

    @Override
    public void onMessage(Order order) {
        log.info("用户微服务收到了订单信息：{}", JSONObject.toJSONString(order));
    }
}

@RocketMQMessageListener 注解，表示当前类是一个 RocketMQ 的消费者，在@RocketMQMessageListener 注解中配置了消费者组为 user-group，主题为 order-topic。

4.2、订单微服务整合 RocketMQ

1、在订单微服务 shop-order 的 pom.xml 文件中添加 RocketMQ 的依赖

2、在订单微服务 shop-order 的 application.yml 文件中添加如下配置：

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: order-group

3、修改 OrderServiceImpl：

@Slf4j
@Service
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements OrderService {
   // ...

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveOrder(OrderParamVo orderParamVo) {
        // ...
        log.info("库存扣减成功");
        // 发送消息
        rocketMQTemplate.convertAndSend("order-topic", order);
    }
}

4.3、测试集成的 RocketMQ

1、分别启动 Nacos，Sentinel，ZipKin 和 RocketMQ

2、分别启动用户微服务、商品微服务、订单微服务和网关服务

3、在浏览器中输入localhost:10001/server-order/order/submit_order?userId=1001&productId=1001&count=1

4、查看用户微服务 shop-user 的控制台，发现会输出订单的信息，如下所示：