正文

必须先理解的rocketmq入门手册，才能再次深入解读(代码片段)

ccwwlx  ccwwlx  2023-03-22  690

关键词：

RocketMQ入门手册

RocketMQ是一个分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点，

同时，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。

具有以下特点：

能够保证严格的消息顺序
提供丰富的消息拉取模式
高效的订阅者水平扩展能力
实时的消息订阅机制
亿级消息堆积能力

RocketMQ 架构原理分析

RocketMQ 架构

技术图片

NameServer （名称服务器）：

提供轻量级的服务发现和路由。NameServer接受来自Broker群集的注册，并提供检测信号机制以检查Broker是否还存在
每个NameServer记录完整的路由信息（Broker 相关 Topic 等元信息，并给 Producer 提供 Consumer 查找 Broker 信息），提供相应的读写服务。

Broker（消息服务器）： 消息存储中心，接收来自 Producer 的消息并存储， Consumer 从这里取得消息

单个Broker节点与所有的NameServer节点保持长连接及心跳，并会定时将Topic信息注册到NameServer，（其底层通信是基于Netty实现的）
Broker负责消息存储，以Topic为维度支持轻量级的队列，单机可以支撑上万队列规模，支持消息推拉模型。
具有上亿级消息堆积能力，同时可严格保证消息的有序性

Producer （生产者）：

负责产生消息，生产者向消息服务器发送由业务应用程序系统生成的消息
生产者支持分布式部署。分布式生产者通过多种负载平衡模式将消息发送到Broker集群。发送过程支持快速失败并且延迟低
三种方式发送消息：同步、异步和单向

Consumer（消费者）：

负责消费消息，消费者从消息服务器拉取信息并将其输入用户应用程序
也支持“推和拉”模型中的分布式部署。
它还支持集群使用和消息广播。它提供了实时消息订阅机制，可以满足大多数消费者的需求。

Broker Server

Broker Server负责消息的存储和传递，消息查询，HA高可用等，Broker Server几个主要模块组成：

技术图片

Remoting Module（远程模块）：broker入口，处理来自客户端的请求

Client Manager（客户端管理）：管理client（生产者/消费者）并维护消费者的主题订阅

Store Service（存储服务）：提供简单的API中数据库中存储或查询消息

HA Service（高可用服务）：提供master broker和slave broker之间的数据同步功能

Index Service（索引服务）：将message建立索引来提供快速的查询能力

RocketMQ 整体流程

启动 NameServer，NameServer启动后进行端口监听，等待 Broker、Producer、Consumer 连上来，相当于一个路由控制中心
Broker 启动，跟所有的 Namesrv 保持长连接，定时发送心跳包
- 心跳包中，包含当前 Broker 信息(IP+端口等)以及存储所有 Topic 信息
- 注册成功后，Namesrv 集群中就有 Topic 跟 Broker 的映射关系
收发消息前，先创建 Topic 。创建 Topic 时，需要指定该 Topic 要存储在哪些 Broker上。也可以在发送消息时自动创建Topic
Producer 发送消息
- 启动时，先跟 Namesrv 集群中的其中一台建立长连接，并从Namesrv 中获取当前发送的 Topic 存在哪些 Broker 上
- 然后跟对应的 Broker 建立长连接，直接向 Broker 发消息
Consumer 消费消息
- 跟其中一台 Namesrv 建立长连接，获取当前订阅 Topic 存在哪些 Broker 上
- 然后直接跟 Broker 建立连接通道，开始消费消息*RocketMQ的消息领域模型

RocketMQ Message

技术图片

Topic（主题）： 表示消息的第一级类型，是最细粒度的订阅单位（生产者传递消息和消费者提取消息标识）

一条消息必须有一个Topic
一个Group可以订阅多个Topic的消息
Topic一般为领域范围，比如交易消息

Tag（标签）： 表示消息的第二级类型，可以是使用相同的Topic不同的Tag来表示同一业务模块的不同任务的消息，比如交易消息又可以分为：交易创建消息，交易完成消息等

助于保持代码整洁和一致
简化RocketMQ提供的查询系统

Message（消息体）： 消息是要传递的信息。 Message中必须包含一个Topic，可选Tag和key-vaule键值对

Message Queue（消息队列）： 所有消息队列都是持久化

一个Topic下可以有多个Queue
Queue的引入使得消息的存储可以分布式集群化，具有了水平扩展能力

Group（组）： 分为Producer Group（生产者组）和Consumer Group（消费者组），具有相同角色组成Group

原生产者在交易后崩溃，broker可以联系同一生产者组的不同生产者实例以进行提交或回退交易。
消费者组的消费者实例必须具有完全相同的主题订阅

RocketMQ 特性

Message Model（消息模式）：

Clustering（集群式）：当使用集群消费模式时，MQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可
Broadcasting（广播式）：当使用广播消费模式时，MQ 会将每条消息推送给集群内所有注册过的客户端，保证消息至少被每台机器消费一次

Message Order（消息顺序）

使用DefaultMQPushConsumer时，可以决定按顺序或同时使用消息
- Orderly：有序地使用消息意味着消息的消费顺序与生产者为每个消息队列发送消息的顺序相同。（如果要处理必须强制执行全局顺序的情况，请确保您使用的主题只有一个消息队列）
如果指定按顺序使用，则消息使用的最大并发度是使用者组订阅的消息队列数
- Concurrently：同时使用消息时，消息使用的最大并发性仅受为每个使用方客户端指定的线程池限制
在此模式下不再保证消息顺序

Message Types（消息类型）

事务消息
顺序消息
延迟消息

RocketMQ单机版安装

下载编译源码

  # 下载$ 
  > wget wget http://mirror.bit.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-source- > 
  # 解压$
  >unzip rocketmq-all-4.7.0-source-release.zip
  > cd rocketmq-all-4.7.0/
  # 编译$
  > mvn -Prelease-all -DskipTests clean install -U
  > cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0

启动 Name Server

 # 启动 Name Server 服务
 > nohup sh bin/mqnamesrv &
 # 启动完成后，查看日志$
 > tail -f ~/logs/rocketmqlogs/namesrv.log
  The Name Server boot success...

启动 Broker

在 conf 目录下，RocketMQ 提供了多种 Broker 的配置文件：
- broker.conf ：单主，异步刷盘。
- 2m/ ：双主，异步刷盘。
- 2m-2s-async/ ：两主两从，异步复制，异步刷盘。
- 2m-2s-sync/ ：两主两从，同步复制，异步刷盘。
- dledger/ ：Dledger 集群，至少三节点
```
 # 启动 Broker服务
 > nohup sh bin/mqbroker -n localhost:9876 &
 # 启动完成后，查看日志$
 > tail -f ~/logs/rocketmqlogs/broker.log 
  The broker[%s, 172.30.30.233:10911] boot success...
```
其中，参数：
- 通过 -c 参数，配置读取的主 Broker 配置
- 通过 -n 参数，设置 RocketMQ Namesrv 地址
Send & Receive Messages（消息发送与接收）

在发送/接收消息之前，我们需要告知client（生产者/消费者）Name Servers的地址。 RocketMQ提供了多种方法来实现：
- 在代码中设置：producer.setNamesrvAddr("ip:port")
- java属性配置：rocketmq.namesrv.addr
- 环境变量配置：NAMESRV_ADDR
- HTTP Endpoint
为简单起见，我们使用环境变量：NAMESRV_ADDR，如下所示：
```
 # 设置 Name Servers的地址$
 > export NAMESRV_ADDR=localhost:9876
 # 生产消息$
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
 SendResult [sendStatus=SEND_OK, msgId= ...
 # 消费消息$ 
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
 ConsumeMessageThread_%d Receive New Messages: [MessageExt...
```

各位看官还可以吗？喜欢的话，动动手指点个??，点个关注呗！！谢谢支持！

欢迎关注公众号【Ccww技术博客】，原创技术文章第一时间推出

技术图片

深入理解rocketmq是如何做到高性能的？

1、RocketMQ的核心Broker对rocketmq稍有了解的同学，都知道它主要由4部分组成，Producer、Consumer、Broker、NameServer。Broker作为RocketMQ的核心，提供了强大的数据存储能力，可以把亿万级的消息存储在服务器磁盘上。它决定... 查看详情

简单理解rocketmq---入门

简介： RocketMQ是阿里巴巴出品的一款低延迟，可靠，可扩展，易于使用的面向消息的中间件，支撑着阿里巴巴集团庞大的消息业务增长。RocketMQ是基于MetaQ的一个开源分支，几乎重写了MetaQ所有的核心组件，可以... 查看详情

深入理解rocketmq普通消息和顺序消息使用，原理，优化(代码片段)

...以这方面为主。这次打压的过程中收获比较的大的是，对RocketMq的一些优化。最开始我们公司使用的是RabbitMq,在一些流量高峰的场景下，发现队列堆积比较严重，导致RabbitMq挂了。为了应对这个场景，最终我们引入了阿里云的Rocke... 查看详情

深入理解rocketmq---实战（控制台搭建）(代码片段)

rocketMQ控制台搭建（1）下载rocketmq-console代码:https://github.com/875279177/incubator-rocketmq-externals（2）修改配置application配置文件，主要修改端口号及rocketmq.config.dataPathserver.contextPath=server.port=8080#spring.ap 查看详情

深入理解java：注解（annotation）自定义注解入门

...a/archive/2013/04/24/3036689.html 　要深入学习注解，我们就必须能定义自己的注解，并使用注解，在定义自己的注解之前，我们就必须要了解Java为我们提供的元注解和相关定义注解的语法。元注解：　　元注解的作用就是负责注... 查看详情

深入理解java：注解（annotation）自定义注解入门

要深入学习注解，我们就必须能定义自己的注解，并使用注解，在定义自己的注解之前，我们就必须要了解Java为我们提供的元注解和相关定义注解的语法。元注解：　　元注解的作用就是负责注解其他注解。Java5.0定义了4个标准... 查看详情

深入理解java的注解（annotation）：自定义注解入门

rocketmq-如何实现顺序消息

...新增消息、修改消息。如何发送和消费顺序消息我们使用RocketMQ顺序消息来模拟一下订单的场景，顺序消息分为两部分：顺序发送、顺序消费。1.顺序发消息上面代码模拟了按顺序依次发送创建、支付、退款消息到TopicTest中。在ap... 查看详情

深入理解java：注解（annotation）自定义注解入门

...peida/archive/2013/04/24/3036689.html要深入学习注解，我们就必须能定义自己的注解，并使用注解，在定义自己的注解之前，我们就必须要了解Java为我们提供的元注解和相关定义注解的语法。元注解：　　元注解的作... 查看详情

分布式事务之深入理解什么是2pc3pc及tcc协议？

导读在上一篇文章《【分布式事务】基于RocketMQ搭建生产级消息集群？》中给大家介绍了基于RocketMQ如何搭建生产级消息集群。因为本系列文章最终的目的是介绍基于RocketMQ的事物消息来解决分布式系统中的数据一致性问题，... 查看详情

深入理解rocketmq---实战（控制台搭建）(代码片段)

spring入门篇

...OP（配置、注解、AspectJ、API）如何学习Spring掌握用法深入理解不断实践反复总结再次深入理解与实践似乎是学习所有技术的方式。。。学习Spring的资源http://spring.io/http://projects.spring.io/spring-framework/Spring是什么&n 查看详情

深入理解rocketmq--消息存储

一、MQ存储分类文件系统：RocketMQ/Kafka/RabbitMQ关系型数据库DB：ActiveMQ（默认采用的KahaDB做消息存储）可选用JDBC的方式来做消息持久化分布式KV存储：ZeroMQ对比：存储效率，文件系统>分布式KV存储>关系型数据库DB易于实现... 查看详情

大家心心念念的rocketmq5.x入门手册来喽

大家好，我是威哥，《RocketMQ实战》作者、RocketMQ社区首席布道师、极客时间《中间件核心技术与实战》专栏作者、中通快递基础架构资深架构师，越努力越幸运，唯有坚持不懈，与大家共勉。1、前言为了更好... 查看详情

rocketmq通信协议

...章目录通讯协议HeaderdatacodelanguageversionopaqueflagremarkextFieldsRocketMQ的通信协议其实很简单，但是无论是官方的用户手册，还是网上的博客，并没有很清晰简单地把其中所有的内容和原理讲明白。对于需要扩展其他语言SDK的开发来说... 查看详情

rocketmq入门到精通—rocketmq初级特性能力|messageorder，rocketmq的消息可以是有序的哦

...源经历了6个月的失踪，我将带着干货终究归来！【RocketMQ入门到精通】订阅与发布消息的发布是指某个生产者向某个topic发送消息，消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息，进而从该topic消费数据... 查看详情

大家心心念念的rocketmq5.x入门手册来喽(代码片段)

1、前言为了更好的拥抱云原生，RocketMQ5.x架构进行了大的重构，提出了存储与计算分离的设计架构，架构设计图如下所示：RocketMQ5.x提供了一套非常建议的消息发送、消费API，并统一放在Apache顶级开源项目rocket... 查看详情