文章目录

简介
下载
- 下载目录地址
安装
- 部署环境要求
- 下载二进制包解压即可
- 启动 NameServer
启动Broker+Proxy
- 单组节点单副本模式启动
使用Java客户端发布订阅消息
- 1. 创建主题 topic
- 2. 创建Java工程使用Maven引入Java SDK包
- 生产者代码 ProducerDemo
- 消费者代码 ConsumerDemo
RocketMQ 的一些基本概念
- NameServer 名字服务器
- Broker 代理服务器（直译为经纪人）
- Proxy （直译为代理）
- Topic 主题
- MessageType 消息类型
- MessageQueue 消息队列
- 消息位点（Offset）
- 消费位点（ConsumerOffset）
- Message 消息
- MessageView 消息视图
- MessageTag 消息标签
- MessageKey 消息索引（或者说消息的关键字）
- 订阅关系（Subscription）
RocketMQ 领域模型
远程通信方式
- 直接远程调用
- 中间件异步调用
消息的传输模型
- 点对点
- 发布订阅

简介

Apache RocketMQ 自诞生以来，因其架构简单、业务功能丰富、具备极强可扩展性等特点被众多企业开发者以及云厂商广泛采用。历经阿里十余年的大规模场景打磨，RocketMQ 已经成为业内共识的金融级可靠业务消息首选方案，被广泛应用于互联网、大数据、移动互联网、物联网等领域的业务场景。

下载

RocketMQ 的安装包分为两种

二进制包：二进制包是已经编译完成后可以直接运行的
源码包：源码包是需要编译后运行的

下载目录地址

https://rocketmq.apache.org/zh/download/
在这里插入图片描述
当前最新的版本为 5.1.3

我们通过二进制下载来安装。

安装

部署环境要求

64位操作系统，推荐 Linux/Unix/macOS
64位 JDK 1.8+

我们的示例在linux CentOS 发行版下安装

下载二进制包解压即可

解压后文件夹名字可能太长，可以修改一下，如我们修改为：rocketmq-5.1.3

如果是源码包，则需要执行如下操作

$ unzip rocketmq-all-5.1.3-source-release.zip
$ cd rocketmq-all-5.1.3-source-release/
$ mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
$ cd distribution/target/rocketmq-5.1.3/rocketmq-5.1.3

启动 NameServer

### 启动namesrv
$ nohup sh bin/mqnamesrv &
 
### 验证namesrv是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log

日志打印出如下字样则说明启动成功
..... The Name Server boot success.  .......

在这里插入图片描述

NameServer 的默认端口为9876，IP 为部署服务器IP地址

启动Broker+Proxy

在 5.0 版本中 Proxy 和 Broker 根据实际诉求可以分为 Local 模式和 Cluster 模式。官网推荐模式为 Local 模式。

在 Local 模式下，Broker 和 Proxy 是同进程部署，只是在原有 Broker 的配置基础上新增 Proxy 的简易配置就可以运行。
在 Cluster 模式下，Broker 和 Proxy 分别部署，即在原有的集群基础上，额外再部署 Proxy 即可。

这两种模式，又对应了多种可部署的组合方式，这个我们后续章再说。我们当前先使用单组节点单副本模式启动。

单组节点单副本模式启动

### 先启动broker  
## -n 后面是 nameServe 的地址，由于我们的broker和nameServer在同一台服务器上执行，所以使用localhost
$ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &

### 验证broker是否启动成功, 比如, broker的ip是192.168.1.2 然后名字是broker-a
$ tail -f ~/logs/rocketmqlogs/proxy.log 

#---- 日志中出现如下内容 broker + proxy 即启动成功。
The broker[broker-a,192.169.1.2:10911] boot success...

broker 默认端口为10911，proxy 默认端口为8081

启动后在 ~/logs/rocketmqlogs 目录下还可以看见 broker.log、namesrv.log、proxy.log

使用Java客户端发布订阅消息

在使用本示例时，很多概念都还没有讲解，对于初次接触的人来说有些困难，这不要紧，直接无脑按顺序操作即可，先体验结果，后续将逐步详细讲解这些内容。

1. 创建主题 topic

5.0版本下创建主题操作，官网推荐使用 mqadmin 命令工具来创建。

$ sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster

在这里插入图片描述

-n 表示 nameserver 的地址，-t 表示主题名称（用于唯一标识主题），-c 集群名称，RocketMQ 默认集群名称就是 DefaultCluster

2. 创建Java工程使用Maven引入Java SDK包

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>${rocketmq-client-java-version}</version>
</dependency>

${rocketmq-client-java-version} 使用最新的版本即可。在maven 仓库中搜索即可。本文编写时的最新版本为：5.0.5。

生产者代码 ProducerDemo

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;

import java.io.IOException;

public class ProducerDemo {

    public static void main(String[] args) throws ClientException, InterruptedException {
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        String endpoint = "192.168.1.1:8081";
        // 消息发送的目标Topic名称，需要提前创建。
        String topic = "TestTopic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();
        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        for(int i = 0; i < 1000;i++) {
            // 普通消息发送。
            Message message = provider.newMessageBuilder()
                    .setTopic(topic)
                    // 设置消息索引键，可根据关键字精确查找某条消息。
                    .setKeys("messageKey")
                    // 设置消息Tag，用于消费端根据指定Tag过滤消息。
                    .setTag("messageTag")
                    // 消息体。
                    .setBody(("messageBody" + i).getBytes())
                    .build();
            try {
                // 发送消息，需要关注发送结果，并捕获失败等异常。
                SendReceipt sendReceipt = producer.send(message);
                System.out.println("Send message successfully, messageId=" + sendReceipt.getMessageId());
            } catch (ClientException e) {
                System.out.println("Failed to send message");
            }
            // 等待10秒再次发送
            Thread.sleep(10000);
        }


        // 发送完，关闭
        try {
            producer.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

控制台会打印类似： Send message successfully, messageId=01005056C0000846DC04DFC51400000000 的结果

消费者代码 ConsumerDemo

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

import java.nio.ByteBuffer;
import java.util.Collections;

public class ConsumerDemo {

    public static void main(String[] args) throws InterruptedException, ClientException {

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        String endpoints = "192.168.1.1:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // 为消费者指定所属的消费者分组。
        String consumerGroup = "YourConsumerGroup";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        String topic = "TestTopic";
        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // 设置消费者分组。
                .setConsumerGroup(consumerGroup)
                // 设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // 设置消费监听器。
                .setMessageListener(messageView -> {
                    System.out.println(messageView);
                    ByteBuffer rs = messageView.getBody();
                    byte[] rsByte = new byte[rs.limit()];
                    rs.get(rsByte);
                    System.out.println("Message body：" + new String(rsByte));
                    // 处理消息并返回消费结果。
                    System.out.println("Consume message successfully, messageId=" + messageView.getMessageId());;
                    return ConsumeResult.SUCCESS;
                })
                .build();

        // 如果不需要再使用 PushConsumer，可关闭该实例。
        // 不调用close 该程序不会关闭
        // pushConsumer.close(); 

    }

}

消费者有3中类型，这里的消费者我们使用了 PushConsumer 。后续我们会详解。打印结果如下：

MessageViewImpl{messageId=01005056C00008017404E0A8AD00000000, topic=TestTopic, bornHost=xxx, bornTimestamp=1691291309533, endpoints=ipv4:192.168.1.1:8081, deliveryAttempt=1, tag=messageTag, keys=[messageKey], messageGroup=null, deliveryTimestamp=null, properties={}}
Message body：messageBody0
Consume message successfully, messageId=01005056C00008017404E0A8AD00000000

RocketMQ 的一些基本概念

NameServer 名字服务器

NameServer是一个简单的 Topic 路由注册中心，支持 Topic、Broker 的动态注册与发现。

主要包括两个功能：

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

NameServer通常会有多个实例部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，客户端仍然可以向其它NameServer获取路由信息。

Broker 代理服务器（直译为经纪人）

Broker主要负责消息的存储、投递和查询以及服务高可用保证。

NameServer几乎无状态节点，因此可集群部署，节点之间无任何信息同步。Broker部署相对复杂。

Proxy （直译为代理）

在官网5.0的文档中NameServer、Broker、Proxy都没有做说明，以上关于NameServer、Broker的说明是4.x的文档说明，（或许是5.0版本2022.9.22才发布，文档没有跟上吧）。

由于没有官方说明，这里只能大胆猜测一下 Proxy 的作用，应该是替代NameServer 和 broker 的一些功能，按照我们示例的写法，以及英文直译它的作用应该是代理网络连接、以及集群下的寻址等功能等。

Topic 主题

RocketMQ 中消息传输和存储的顶层容器，用于标识同一类业务逻辑的消息。主题通过TopicName来做唯一标识和区分。一般我们定义不同的主题来区分不同的业务逻辑。

MessageType 消息类型

RocketMQ 中按照消息传输特性的不同而定义的分类，用于类型管理和安全校验。 RocketMQ 支持的消息类型有普通消息、顺序消息、事务消息和定时/延时消息。

从5.0版本开始，支持强制校验消息类型，即每个主题Topic只允许发送一种消息类型的消息，这样可以更好的运维和管理生产系统，避免混乱。但同时保证向下兼容4.x版本行为，强制校验功能默认关闭，推荐通过服务端参数 enableTopicMessageTypeCheck 手动开启校验。

MessageQueue 消息队列

队列是 RocketMQ 中消息存储和传输的实际容器，也是消息的最小存储单元。通过QueueId来做唯一标识和区分。RocketMQ 的所有主题都是由多个队列组成。任意一个消息队列在逻辑上都是无限存储，即消息位点会从0到Long.MAX无限增加。

消息位点（Offset）

消息是按到达服务端的先后顺序存储在指定主题的多个队列中，每条消息在队列中都有一个唯一的Long类型坐标，这个坐标被定义为消息位点。

队列中最早一条消息的位点为最小消息位点（MinOffset）；最新一条消息的位点为最大消息位点（MaxOffset）。

消费位点（ConsumerOffset）

RocketMQ 通过消费位点管理消息的消费进度。每条消息被某个消费者消费完成后不会立即在队列中删除，RocketMQ 会基于每个消费者分组维护一份消费记录，该记录指定消费者分组消费某一个队列时，消费过的最新一条消息的位点，即消费位点。

Message 消息

消息是 RocketMQ 中的最小数据传输单元。生产者将业务数据的负载和拓展属性包装成消息发送到服务端，服务端按照相关语义将消息投递到消费端进行消费。

MessageView 消息视图

消息视图是 RocketMQ 面向开发视角提供的一种消息只读接口。通过消息视图可以读取消息内部的多个属性和负载信息，但是不能对消息本身做任何修改。

MessageTag 消息标签

消息标签是 RocketMQ 提供的细粒度消息分类属性，可以在主题层级之下做消息类型的细分。消费者通过订阅特定的标签来实现细粒度过滤。比如：我们一般将标签设计为业务操作级，如果消息都是下单成功的消息，那么他们的标签为“orderSuccess”，相同标签的消息，调用相同的业务逻辑。这样我们可以减少要创建的主题的个数。