Spring Cloud Alibaba 实战（六）RocketMQ篇

刺骨的言语ヽ痛彻心扉 2022-09-04 11:53 216阅读 0赞

# 1. RocketMQ 简介 #

> RocketMQ 是阿里巴巴开源的分布式消息中间件。支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。它里面有几个区别于标准消息中件间的概念，如Group、Topic、Queue等。系统组成则由Producer、Consumer、Broker、NameServer等。

## RocketMQ 特点 ##

*  支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型；
 *  在一个队列中可靠的先进先出（FIFO）和严格的顺序传递，RocketMQ 可以保证严格的消息顺序，而ActiveMQ 无法保证；
 *  支持拉（Pull）和推（Push）两种消息模式；Push 好理解，比如在消费者端设置 Listener 回调；而 Pull，控制权在于应用，即应用需要主动的调用拉消息方法从 Broker 获取消息，这里面存在一个消费位置记录的问题（如果不记录，会导致消息重复消费）；
 *  单一队列百万消息的堆积能力；RocketMQ 提供亿级消息的堆积能力，这不是重点，重点是堆积了亿级的消息后，依然保持写入低延迟；
 *  支持多种消息协议，如 JMS、MQTT 等；
 *  分布式高可用的部署架构，满足至少一次消息传递语义；RocketMQ 原生就是支持分布式的，而ActiveMQ 原生存在单点性；
 *  提供 docker 镜像用于隔离测试和云集群部署；
 *  提供配置、指标和监控等功能丰富的 Dashboard。

## Broker ##

> Broker 其实就是 RocketMQ 服务器，负责存储消息、转发消息。Broker 在 RocketMQ 系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。Broker 也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。

Broker Server 是 RocketMQ 真正的业务核心，包含了多个重要的子模块：

*  **路由模块**：整个 Broker 的实体，负责处理来自 clients 端的请求。
 *  **客户端管理**：负责管理客户端(Producer/Consumer)和维护 Consumer 的 Topic 订阅信息
 *  **存储服务**：提供方便简单的 API 接口处理消息存储到物理硬盘和查询功能。
 *  **高可用服务**：高可用服务，提供 Master Broker 和 Slave Broker 之间的数据同步功能。
 *  **消息索引服务**：根据特定的 Message key 对投递到 Broker 的消息进行索引服务，以提供消息的快速查询。

## NameServer ##

> NameServer 是一个非常简单的 Topic 路由注册中心，其角色类似 Dubbo 中的 zookeeper，支持 Broker 的动态注册与发现。

主要包括两个功能：

*  **Broker 管理**：NameServer 接受 Broker 集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查 Broker 是否还存活；
 *  **路由信息管理**：给 Producer 和 Consumer 提供服务获取 Broker 列表。每个 NameServer 将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。然后 Producer 和 Conumser 通过 NameServer 就可以知道整个 Broker 集群的路由信息，从而进行消息的投递和消费。

# 2. 使用 Docker 快速搭建 RocketMQ 4.4 #

rocketmq 需要部署 broker 与 nameserver ，考虑到分开部署比较麻烦，这里将会使用 docker-compose。另外，还需要搭建一个 web 可视化控制台，可以监控 mq 服务状态，以及消息消费情况，这里使用 rocketmq-console，同样该程序也将使用 docker 安装。

1.  **在 linux 服务器上选择并建立目录**；

mkdir rocketmq-docker

1.  **进入 rocketmq-docker 目录，建立一个名为 broker.conf 的配置文件，内容如下**：

# 所属集群名称，如果节点较多可以配置多个
    brokerClusterName = DefaultCluster
    # broker名称，master和slave使用相同的名称，表明他们的主从关系
    brokerName = broker-a
    # 0表示Master，大于0表示不同的slave
    brokerId = 0
    # 表示几点做消息删除动作，默认是凌晨4点
    deleteWhen = 04
    # 在磁盘上保留消息的时长，单位是小时
    fileReservedTime = 48
    # 有三个值：SYNC_MASTER，ASYNC_MASTER，SLAVE；同步和异步表示Master和Slave之间同步数据的机制；
    brokerRole = ASYNC_MASTER
    # 刷盘策略，取值为：ASYNC_FLUSH，SYNC_FLUSH表示同步刷盘和异步刷盘；SYNC_FLUSH消息写入磁盘后才返回成功状态，ASYNC_FLUSH不需要；
    flushDiskType = ASYNC_FLUSH
    # 设置broker节点所在服务器的ip地址
    # brokerIP1 = 192.168.138.131

**注意**：这里的 brokerIP1 请务必设置，否则默认会成为 docker 容器内部IP导致外网链接不上。

1.  **还是在 rocketmq-docker 目录，建立一个名为 rocketmq.yaml 的脚本文件**；

![图片][fa78bfc793c4e7eaff04697e23be6255.png]

rocketmq.yaml 内容如下:

version: '2'
    services:
      namesrv:
        image: rocketmqinc/rocketmq
        container_name: rmqnamesrv
        ports:
          - 9876:9876
        volumes:
          - /docker/rocketmq/data/namesrv/logs:/home/rocketmq/logs
          - /docker/rocketmq/data/namesrv/store:/home/rocketmq/store
        command: sh mqnamesrv
      broker:
        image: rocketmqinc/rocketmq
        container_name: rmqbroker
        ports:
          - 10909:10909
          - 10911:10911
          - 10912:10912
        volumes:
          - /docker/rocketmq/data/broker/logs:/home/rocketmq/logs
          - /docker/rocketmq/data/broker/store:/home/rocketmq/store
          - /docker/rocketmq/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf
        command: sh mqbroker -n namesrv:9876 -c /opt/rocketmq-4.4.0/conf/broker.conf
        depends_on:
          - namesrv
        environment:
          - JAVA_HOME=/usr/lib/jvm/jre
      console:
        image: styletang/rocketmq-console-ng
        container_name: rocketmq-console-ng
        ports:
          - 8087:8080
        depends_on:
          - namesrv
        environment:
          - JAVA_OPTS= -Dlogging.level.root=info   -Drocketmq.namesrv.addr=rmqnamesrv:9876 
          - Dcom.rocketmq.sendMessageWithVIPChannel=false

1.  **开启 broker 用到的防火墙端口，方便后续使用**：

firewall-cmd --zone=public --add-port=10909-10912/tcp --permanent

1.  **执行 sentinel-dashboard.yaml 脚本启动容器**：

docker-compose -f rocketmq.yaml up

1.  **进入 rocketmq 控制台，我们会发现类似的图表（当然开始肯定是空的）**：

http://(安装RocketMQ机器的IP):8087

![20210813171406151.gif][]

1.  **我们选择 “集群” 这一栏，进入就能看见我们设置的 broker 的外网IP了**；

![20210813171406151.gif][]

到这里我们理论上已经完成 RocketMQ 服务端的部署，现在就可以去 Spring 项目中使用客户端了。

# 3. 在 Spring 项目中引入 RocketMQ 客户端 #

1.  **添加 pom 文件依赖**：

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

1.  **在 application.yml 添加配置**：

server:
      port: 10801
      
    spring:
      application:
        name: (项目名称)-service
        
    rocketmq:
      name-server: (安装RocketMQ机器的IP):9876
      producer:
        group: (项目名称)-group

1.  **新建一个消息发送类 MessageProducer 作为消息的 生产者**：

@Service
    public class MessageProducer implements CommandLineRunner {
    
        @Resource
        private RocketMQTemplate rocketMQTemplate;
    
        @Override
        public void run(String... args) throws Exception {
            rocketMQTemplate.send("test-topic-1", MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
        }
    
    }

1.  **新建一个消息接收类 MessageListener 作为消息的 消费者**：

@Slf4j
    @Service
    @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
    public class MessageListener implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String message) {
            log.info("received message: {}", message);
        }
    
    }

1.  **新建一个 MessageProducer 的调用控制器类**：

@RestController
    @RequestMapping
    public class HelloController {
    
        @Resource
        private MessageProducer messageProducer;
    
        @RequestMapping("/message")
        public void message() throws Exception {
            messageProducer.run("");
        }

1.  **启动 Spring 项目，我们来测试一下最简单的消息发送**:

GET http://localhost:10801/message
    Accept: */*
    Cache-Control: no-cache

![20210813171406150.gif][]

我们也能在 RocketMQ 的管理控制台，大致看到消息的消费情况：

![20210813171406150.gif][]

实战到了这里并没有解释一些概念，可能大家会有点犯迷糊。

> Topic（主题）相当于一种类型的消息，比如可以设置 Topic1 专门是分销的业务， Topic2 专门是优惠券的业务；而 Group（分组）相当于对生产者和消费者的分组，那我们的微服务既能做生产者也能做消费者，比如可以设置 Group1 是商品的微服务，Group2 是订单的微服务，当然还要区分下是生产者分组还是消费者分组。

*  **Group**：分为ProducerGroup 和 ConsumerGroup, 代表某一类的生产者和消费者，一般来说同一个服务可以作为 Group，同一个 Group 一般来说发送和消费的消息都是一样的。
 *  **Topic**：消息主题，一级消息类型，生产者向其发送消息，消费者读取其消息。
 *  **Queue**: 分为读和写两种队列，一般来说读写队列数量一致，如果不一致就会出现很多问题。

**Topic 中有分为了多个 Queue，这其实是我们发送/读取消息通道的最小单位**。我们发送消息都需要指定某个写入某个 Queue，拉取消息的时候也需要指定拉取某个 Queue，所以我们的顺序消息可以基于我们的 Queue 维度保持队列有序，如果想做到全局有序那么需要将 Queue 大小设置为1，这样所有的数据都会在 Queue 中有序。

[fa78bfc793c4e7eaff04697e23be6255.png]: /images/20220829/939f1b59dcb243faac56d0eb39a70f39.png
[20210813171406151.gif]: /images/20220829/1acb629be7984d61948b37fbee123c1e.png
[20210813171406150.gif]: /images/20220829/3496fe2994e44e4e9d789ae99bb600b9.png