Rocketmq特性详解&场景介绍

左手的ㄟ右手 2022-04-23 02:18 152阅读 0赞

# Rocketmq特性详解&场景介绍 #

# 整体认识rocketmq #

## 架构： ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 1][]

**整体认识：**远程通讯，发送消息，存储消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 2][]

## 概念： ##

**Producer**：消息生产者，负责产生消息，一般由业务系统负责产生消息

**Consumer**：消息消费者，负责消费消息，一般是后台系统负责异步消费

**Topic****：**消息主题，负责标记一类消息，生产者将消息发送到Topic，消费者从该Topic消费消息

**Broker****：**消息中转角色，负责存储消息，转发消息，一般也称为 Server，在 JMS 规范中称为 Provider

**NameServer****：**服务发现Server，用于生产者和消费者获取Broker的服务；

## Rocketmq模块划分： ##

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 名称 </td> 
 <td style="vertical-align:top;width:213.05pt;"> 作用 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> broker </td> 
 <td style="vertical-align:top;width:213.05pt;"> broker模块：c和p端消息存储逻辑 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> client </td> 
 <td style="vertical-align:top;width:213.05pt;"> 客户端api：produce、consumer端 接受与发送api </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> common </td> 
 <td style="vertical-align:top;width:213.05pt;"> 公共组件：常量、基类、数据结构 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> tools </td> 
 <td style="vertical-align:top;width:213.05pt;"> 运维tools：命令行工具模块 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> store </td> 
 <td style="vertical-align:top;width:213.05pt;"> 存储模块：消息、索引、commitlog存储 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> namesrv </td> 
 <td style="vertical-align:top;width:213.05pt;"> 服务管理模块：服务注册topic等信息存储 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> remoting </td> 
 <td style="vertical-align:top;width:213.05pt;"> 远程通讯模块：netty+fastjson </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> logappender </td> 
 <td style="vertical-align:top;width:213.05pt;"> 日志适配模块 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> example </td> 
 <td style="vertical-align:top;width:213.05pt;"> Demo列子 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> filtersrv </td> 
 <td style="vertical-align:top;width:213.05pt;"> 消息过滤器模块 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> srvutil </td> 
 <td style="vertical-align:top;width:213.05pt;"> 辅助模块 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> filter </td> 
 <td style="vertical-align:top;width:213.05pt;"> 过滤模块：消息过滤模块 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> distribution </td> 
 <td style="vertical-align:top;width:213.05pt;"> 部署、运维相关zip包中的代码 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> openmessaging </td> 
 <td style="vertical-align:top;width:213.05pt;"> 兼容openmessaging分布式消息模块 </td> 
 </tr> 
 </tbody> 
</table>

# Rocketmq高可用： #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 3][]

# 特性： #

# Producer端： #

## 发送方式： ##

**Sync****：**同步的发送方式，会等待发送结果后才返回

**Async****：**异步的发送方式，发送完后，立刻返回。Client 在拿到 Broker 的响应结果后，会回调指定的 callback. 这个 API 也可以指定 Timeout，不指定也是默认的 3000ms.

**Oneway****：**比较简单，发出去后，什么都不管直接返回。Ps:日志

## 发送结果： ##

org.apache.rocketmq.client.producer.SendStatus

SEND\_OK,：消息发送成功   
FLUSH\_DISK\_TIMEOUT,消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失   
FLUSH\_SLAVE\_TIMEOUT,消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失   
SLAVE\_NOT\_AVAILABLE,

消息发送成功，但是此时 slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢

## 普通消息： ##

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl\#sendDefaultImpl

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 5][]

1.  准备工作 mesasge、网络相关、线程相关
2.  从namesrv获取topic路由（缓存机制）
3.  组装数据，broker需要的序列化数据（json）
4.  Netty发送（源码）

## 定时消息 ##

定时消息是指消息发到 Broker 后，不能立刻被 Consumer 消费，要到特定的时间点或者等待特定的时间后才能被消费。（第三方 job 步长）

固定精度：

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

org.apache.rocketmq.store.config.MessageStoreConfig\#messageDelayLevel

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 6][]

## 顺序消息： ##

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl\#sendSelectImpl

场景：订单》下单》支付》配送》签收

底层原理：4个队列，一个订单下面不同状态的消息是顺序的只需要发到一个队列中

org.apache.rocketmq.client.producer.MessageQueueSelector 如何选择一个队列

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 7][]

## 事务消息 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 8][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 9][]

事务：ACID xa 二段阶段 三阶段 tcc 阿里

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 10][]

http://rocketmq.apache.org/rocketmq/the-design-of-transactional-message/

# Consumer端： #

## 消费模型： ##

org.apache.rocketmq.common.protocol.heartbeat.MessageModel\#BROADCASTING

org.apache.rocketmq.common.protocol.heartbeat.MessageModel\#CLUSTERING

## 消费选择： ##

org.apache.rocketmq.common.consumer.ConsumeFromWhere\#CONSUME\_FROM\_LAST\_OFFSET

第一次启动从队列**最后**位置消费，后续再启动接着上次消费的进度开始消费

org.apache.rocketmq.common.consumer.ConsumeFromWhere\#CONSUME\_FROM\_FIRST\_OFFSET

第一次启动从队列**初始位置**消费，后续再启动接着上次消费的进度开始消费

org.apache.rocketmq.common.consumer.ConsumeFromWhere\#CONSUME\_FROM\_TIMESTAMP

第一次启动从**指定时间点位置**消费，后续再启动接着上次消费的进度开始消费

以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始

## 消息重复幂等： ##

RocketMQ无法避免消息重复，所以如果业务对消费重复非常敏感，务必要在业务层面去重

Ps：见开发文档

**为什么要有组group：?**

sh mqshutdown broker

sh mqshutdown namesrv

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70]: /images/20220323/5f060a74846641f1be01dbaa7bc3e349.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 1]: /images/20220323/d8d8799127b541d8be536f01dc9f2c11.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 2]: /images/20220323/ccc982712ac3474aae3bdb0d4e29333f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 3]: /images/20220323/3636b370067e41ce8c4bcdbb21e1b1de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 4]: /images/20220323/9ab3a38c2cfd43f5aeebc18836a2431b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 5]: /images/20220323/bf54d9bd17014b11913c1b4794eceebf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 6]: /images/20220323/01e716773cf54a738faa8adfdbc55e35.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 7]: /images/20220323/04b26258bb544e1d833307246818d43a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 8]: /images/20220323/481c6be847b94952b9052bad7efcedd8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 9]: /images/20220323/930e4b3562de4edf93972d4a3335095d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o3ODU0OTUyOTU_size_16_color_FFFFFF_t_70 10]: /images/20220323/989dccd9f26847eb8056ff2ed8ac0f93.png