一、Apache

是一个开源的分布式消息中间件#xff0c;由阿里巴巴集团开发并…本文尝试从Apache

RocketMQ的简介、主要组件及其作用、3种部署模式、Controller集群模式工作流程、最佳实践等方面对其进行详细分析。

希望对您有所帮助

一、Apache

软件基金会。

它旨在提供高吞吐量、低延迟和高可靠性的消息传递和流处理服务。

广泛应用于金融、互联网、物联网等领域支持多种应用场景。

核心特性

高吞吐量支持每秒处理数百万条消息适用于高并发的业务场景。

低延迟具备低于毫秒级的延迟确保消息能够快速传递。

数据一致性通过消息确认和重新传递机制确保消息的可靠传递。

消息持久化支持消息持久化保证在系统故障时消息不丢失。

水平扩展支持通过增加节点的方式水平扩展满足业务增长需求。

弹性伸缩根据业务负载动态调整资源提升资源利用效率。

灵活性

多种消息模型支持点对点、发布/订阅、流处理等多种消息传递模型满足不同应用需求。

丰富的接口提供Java、C、Python等多种语言的客户端接口便于集成和使用。

关键功能和支持特性

集群。

生产者可以指定消息的主题Topic和标签Tag便于消息分类和筛选。

Consumer消费者负责从

RocketMQ

集群接收和处理消息。

消费者分为集群消费和广播消费两种模式。

NameServer命名服务器提供轻量级的路由服务存储生产者和消费者与

Broker

之间的路由信息。

支持动态扩展确保系统的高可用性。

Broker代理服务器负责存储和转发消息。

Broker

将消息存储在磁盘上并根据需要将消息传递给消费者。

支持主从架构提升系统的容错能力和数据可靠性。

事务消息支持分布式事务消息确保在分布式系统中实现最终一致性。

事务消息可以确保消息在事务成功时被发送事务失败时被回滚。

顺序消息支持顺序消息传递确保消息按照发送顺序被消费。

定时消息和延时消息支持消息定时发送和延时发送满足特定的业务需求。

批量消息支持批量消息发送提高消息传递效率。

消息过滤支持基于标签的消息过滤消费者可以根据标签选择性地接收消息。

消息重试支持消息重试机制确保消息在消费失败时可以重新消费。

死信队列DLQ支持死信队列确保处理失败的消息不会丢失。

流量控制支持流量控制防止系统过载。

分布式架构支持多数据中心部署和跨地域部署提升系统的高可用性和容灾能力。

监控和管理提供完善的监控和管理工具支持消息统计、系统监控和运维管理。

安全性支持基于权限的访问控制确保消息传递的安全性。

使用场景

异步通信在分布式系统中实现异步消息传递解耦系统组件提升系统的响应速度和可靠性。

事件驱动架构构建基于事件的系统通过消息队列实现事件驱动的业务逻辑提升系统的灵活性和可维护性。

日志收集集中收集和处理分布式系统中的日志信息实现统一的日志管理和分析。

流处理实时处理和分析数据流支持大数据分析、实时监控等应用场景。

高性能、高吞吐量适用于大规模、高并发的业务场景。

强一致性通过严格的消息确认机制确保数据一致性。

灵活的扩展性支持弹性伸缩能够适应业务的快速变化。

多语言支持提供多种语言的客户端接口便于开发和集成。

Apache

以其高性能、高可靠性和灵活性成为众多企业实现分布式消息传递和流处理的首选解决方案。

二、RocketMQ

的地址列表和路由信息供生产者和消费者查询。

动态注册Broker

启动时向

消息存储持久化存储消息确保数据的可靠性和持久性。

消息转发将消息从生产者传递到消费者。

主从架构支持主从模式主

Broker

备份数据提供容错能力。

消息索引通过建立消息索引提高消息检索效率。

Producer生产者

消息生成创建并发送消息到指定的主题Topic。

异步发送支持异步消息发送提高发送性能。

路由获取通过

NameServer

消息消费从指定的主题Topic接收并处理消息。

消费模式支持集群消费和广播消费两种模式。

集群消费同一消费组内的多个消费者负载均衡地处理消息每条消息只会被一个消费者处理一次。

广播消费每个消费者都会处理所有的消息每条消息会被所有消费者处理一次。

Controller控制器

事务管理负责分布式事务的管理确保消息在事务成功时被发送事务失败时被回滚。

状态监控监控

Broker

等信息。

消息查询支持消息的精确查询和模糊查询方便运维人员排查问题。

配置管理提供

Topic、Consumer

CommitLog存储所有消息的物理文件按顺序写入支持快速写入操作。

ConsumeQueue消息的逻辑队列记录消息在

CommitLog

中的位置便于消费者快速检索。

IndexFile消息索引文件通过消息的属性如消息键建立索引支持快速查询。

Client客户端

等多种语言的客户端接口便于集成和使用。

消息操作支持消息的发送、接收、过滤、批处理等操作。

Apache

通过这些组件的协同工作提供了高性能、高可靠性和高可扩展性的分布式消息传递和流处理服务。

各个组件分工明确确保系统能够高效、稳定地运行并满足各种复杂业务场景的需求。

三、RocketMQ

模式一般情况下如果没有特殊需求或者遵循从早期版本平滑升级的思路可以选用Local模式。

Local

NameServer需要先于Broker启动且如果在生产环境使用为了保证高可用建议一般规模的集群启动3个NameServer各节点的启动命令相同如下

###

~/logs/rocketmqlogs/namesrv.log

The

只有一个节点一旦Broker重启或者宕机时会导致整个服务不可用。

不建议线上环境使用,

BrokerProxy

是否启动成功例如Broker的IP为192.168.1.2且名称为broker-a

tail

~/logs/rocketmqlogs/broker_default.log

The

副本例如2个Master或者3个Master这种模式的优缺点如下

优点配置简单单个Master宕机或重启维护对应用无影响在磁盘配置为RAID10时即使机器宕机不可恢复情况下由于RAID10磁盘非常可靠消息也不会丢异步刷盘丢失少量消息同步刷盘一条不丢性能最高

缺点单台机器宕机期间这台机器上未被消费的消息在机器恢复之前不可订阅消息实时性会受到影响。

启动BrokerProxy集群

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties

--enable-proxy

在机器B启动第二个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties

--enable-proxy

如上启动命令是在单个NameServer情况下使用的。

对于多个NameServer的集群Broker启动命令中-n后面的地址列表用分号隔开即可例如

192.168.1.1:9876;192.161.2:9876。

多节点集群多副本模式-异步复制

Master-SlaveHA采用异步复制方式主备有短暂消息延迟毫秒级这种模式的优缺点如下

优点即使磁盘损坏消息丢失的非常少且消息实时性不会受影响同时Master宕机后消费者仍然可以从Slave消费而且此过程对应用透明不需要人工干预性能同多Master模式几乎一样

启动BrokerProxy集群

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties

--enable-proxy

在机器B启动第二个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-b.properties

--enable-proxy

在机器C启动第一个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties

--enable-proxy

在机器D启动第二个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-b-s.properties

--enable-proxy

Master-SlaveHA采用同步双写方式即只有主备都写成功才向应用返回成功这种模式的优缺点如下

优点数据与服务都无单点故障Master宕机情况下消息无延迟服务可用性与数据可用性都非常高

缺点性能比异步复制模式略低大约低10%左右发送单个消息的RT会略高且目前版本在主节点宕机后备机不能自动切换为主机。

BrokerProxy

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties

--enable-proxy

在机器B启动第二个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties

--enable-proxy

在机器C启动第一个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties

--enable-proxy

在机器D启动第二个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties

--enable-proxy

来区分。

$ROCKETMQ_HOME指的RocketMQ安装目录需要用户自己设置此环境变量。

5.0

提供更具灵活性的HA机制让用户更好的平衡成本、服务可用性、数据可靠性同时支持业务消息和流存储的场景。

详见

五、Cluster模式部署

~/logs/rocketmqlogs/namesrv.log

The

只有一个节点一旦Broker重启或者宕机时会导致整个服务不可用。

不建议线上环境使用,

###

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

副本例如2个Master或者3个Master这种模式的优缺点如下

优点配置简单单个Master宕机或重启维护对应用无影响在磁盘配置为RAID10时即使机器宕机不可恢复情况下由于RAID10磁盘非常可靠消息也不会丢异步刷盘丢失少量消息同步刷盘一条不丢性能最高

缺点单台机器宕机期间这台机器上未被消费的消息在机器恢复之前不可订阅消息实时性会受到影响。

###

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties

###

在机器B启动第二个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties

...如上启动命令是在单个NameServer情况下使用的。

对于多个NameServer的集群Broker启动命令中-n后面的地址列表用分号隔开即可例如

192.168.1.1:9876;192.161.2:9876。

多节点集群多副本模式-异步复制

Master-SlaveHA采用异步复制方式主备有短暂消息延迟毫秒级这种模式的优缺点如下

优点即使磁盘损坏消息丢失的非常少且消息实时性不会受影响同时Master宕机后消费者仍然可以从Slave消费而且此过程对应用透明不需要人工干预性能同多Master模式几乎一样

###

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties

###

在机器B启动第二个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-b.properties

###

在机器C启动第一个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties

###

在机器D启动第二个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-async/broker-b-s.properties

多节点集群多副本模式-同步双写

Master-SlaveHA采用同步双写方式即只有主备都写成功才向应用返回成功这种模式的优缺点如下

优点数据与服务都无单点故障Master宕机情况下消息无延迟服务可用性与数据可用性都非常高

缺点性能比异步复制模式略低大约低10%左右发送单个消息的RT会略高且目前版本在主节点宕机后备机不能自动切换为主机。

###

在机器A启动第一个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties

###

在机器B启动第二个Master例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties

###

在机器C启动第一个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties

###

在机器D启动第二个Slave例如NameServer的IP为192.168.1.1

nohup

$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties

Broker

来区分。

$ROCKETMQ_HOME指的RocketMQ安装目录需要用户自己设置此环境变量。

5.0

提供更具灵活性的HA机制让用户更好的平衡成本、服务可用性、数据可靠性同时支持业务消息和流存储的场景。

详见

Proxy

在机器A启动第一个Proxy例如NameServer的IP为192.168.1.1

nohup

在机器B启动第二个Proxy例如NameServer的IP为192.168.1.1

nohup

在机器C启动第三个Proxy例如NameServer的IP为192.168.1.1

nohup

挂掉多数派只影响切换能力不影响原来路由获取等功能。

另一种是独立部署需要单独部署

Controller

enableControllerInNamesrvtrue并填上

Controller

n0-127.0.0.1:9877;n1-127.0.0.1:9878;n2-127.0.0.1:9879

controllerStorePath

enableControllerInNamesrvNameserver

controller默认

false。

controllerDLegerGroupDLedger

Raft

保持一致即可。

controllerDLegerPeersDLedger

Group

内的各个节点配置必须要保证一致。

controllerDLegerSelfId节点

id必须属于

内各个节点要唯一。

controllerStorePathcontroller

重启或宕机需要依靠日志来恢复数据该目录非常重要不可以轻易删除。

enableElectUncleanMaster是否可以从

SyncStateSet

distribution/bin/mqcontroller配置参数与内嵌模式相同。

独立部署Controller后仍然需要单独部署NameServer提供路由发现能力

Broker

127.0.0.1:9877;127.0.0.1:9878;127.0.0.1:9879syncBrokerMetadataPeriod向

controller

50005s。

checkSyncStateSetPeriod检查

SyncStateSet

SyncState。

默认50005s。

syncControllerMetadataPeriod同步

controller

的地址。

默认1000010s。

haMaxTimeSlaveNotCatchup表示

Slave

副本进行异步复制。

默认为false。

inSyncReplicas需保持同步的副本组数量默认为1allAckInSyncStateSettrue

时该参数无效。

minInSyncReplicas最小需保持同步的副本组数量若

SyncStateSet

PutMessageStatus.IN_SYNC_REPLICAS_NOT_ENOUGH默认为1。

nohup

自动主备切换模式下Broker无需指定brokerId和brokerRole其由Controller组件进行分配

兼容性

enableControllerModefalse则仍然以之前方式运行。

若设置

controller

Broker正常运行无法切换正常运行无法切换正常运行无法切换正常运行无法切换新版

Broker

模式无法正常上线正常运行可以切换正常运行可以切换无法正常上线新版

Broker

模式正常运行无法切换正常运行无法切换正常运行无法切换正常运行无法切换

升级注意事项

不对齐需要保证主上线以后再上线备否则可能会因为数据截断而丢失消息。

DLedger

一段时间只要确认存量消息被全部消费即可比如根据消息的保存时间来决定然后清空

store

config/topics.json、subscriptionGroup.json

下保留

ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

.error-icon{fill:#552222;}#mermaid-svg-6ZiyOaOZMYevYezd

.error-text{fill:#552222;stroke:#552222;}#mermaid-svg-6ZiyOaOZMYevYezd

.edge-thickness-normal{stroke-width:2px;}#mermaid-svg-6ZiyOaOZMYevYezd

.edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-6ZiyOaOZMYevYezd

.edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-6ZiyOaOZMYevYezd

.edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-6ZiyOaOZMYevYezd

.edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-6ZiyOaOZMYevYezd

.marker{fill:#333333;stroke:#333333;}#mermaid-svg-6ZiyOaOZMYevYezd

.marker.cross{stroke:#333333;}#mermaid-svg-6ZiyOaOZMYevYezd

ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-6ZiyOaOZMYevYezd

ms",verdana,arial,sans-serif;color:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

.cluster-label

text{fill:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

.cluster-label

span{color:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

.label

text,#mermaid-svg-6ZiyOaOZMYevYezd

span{fill:#333;color:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

.node

rect,#mermaid-svg-6ZiyOaOZMYevYezd

.node

circle,#mermaid-svg-6ZiyOaOZMYevYezd

.node

ellipse,#mermaid-svg-6ZiyOaOZMYevYezd

.node

polygon,#mermaid-svg-6ZiyOaOZMYevYezd

.node

path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-6ZiyOaOZMYevYezd

.node

.label{text-align:center;}#mermaid-svg-6ZiyOaOZMYevYezd

.node.clickable{cursor:pointer;}#mermaid-svg-6ZiyOaOZMYevYezd

.arrowheadPath{fill:#333333;}#mermaid-svg-6ZiyOaOZMYevYezd

.edgePath

.path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-6ZiyOaOZMYevYezd

.flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-6ZiyOaOZMYevYezd

.edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-6ZiyOaOZMYevYezd

.edgeLabel

rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-6ZiyOaOZMYevYezd

.cluster

rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-6ZiyOaOZMYevYezd

.cluster

text{fill:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

.cluster

span{color:#333;}#mermaid-svg-6ZiyOaOZMYevYezd

div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:"trebuchet

ms",verdana,arial,sans-serif;font-size:12px;background:hsl(80,

100%,

#aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-6ZiyOaOZMYevYezd

:root{--mermaid-font-family:"trebuchet

Broker

多主多备在生产环境中推荐使用多主多备的部署方式。

多个主节点Master和备份节点Slave可以确保在主节点发生故障时备份节点能够快速接管保证系统的高可用性。

跨机房部署将

Broker

配置文件中设置消息过期时间默认是72小时。

过期的消息会被定期清理释放存储空间。

手动清理在必要时可以手动清理过期消息确保存储空间的充足。

性能优化

方法发送批量消息提高发送效率。

批量大小控制合理设置批量消息的大小避免单次发送的数据量过大导致网络拥堵。

消费端并发

参数来调整消费线程数。

顺序消费对于需要保证顺序的消息使用顺序消费模式Orderly避免消息乱序。

异步发送与回调

方法发送消息避免同步等待提高发送性能。

回调处理结合回调函数处理发送结果及时处理发送失败的情况。

消息重试和补偿

参数设置最大重试次数。

重试间隔合理设置重试间隔避免频繁重试导致的资源浪费。

业务补偿机制

幂等性设计在关键业务场景中设计幂等性操作确保多次处理同一消息不会产生副作用。

补偿事务对于分布式事务场景设计补偿机制确保事务的一致性。

监控和报警

详细日志启用详细的日志记录功能记录消息的发送、接收和消费的详细信息。

可以通过调整

logLevel

参数设置日志级别。

日志轮转定期轮转日志文件避免日志文件过大影响系统性能。

报警设置

阈值报警设置合理的报警阈值当系统指标如消息堆积、发送失败率等超过阈值时触发报警。

邮件和短信通知配置报警通知通过邮件和短信及时通知运维人员进行处理。

安全性

协议保护数据的安全。

存储加密对敏感数据进行加密存储确保数据在磁盘上的安全。

网络隔离

集群确保在一个地域发生故障时能够自动切换到其他地域继续提供服务。

数据同步利用

RocketMQ

定期备份定期进行数据备份确保在发生数据丢失时能够快速恢复数据。

可以通过定期备份

CommitLog

数据来实现。

异地备份将备份数据存储在异地防止同一地点的灾难导致备份数据丢失。

RocketMQ

的稳定性、性能和安全性确保消息系统能够高效、可靠地运行。

结合具体的业务需求和系统环境合理配置和优化

九、Go语言实践RocketMQ样例

客户端库来进行消息的生产和消费。

下面是一个简单的实践示例包括如何配置

RocketMQ

github.com/apache/rocketmq-client-go/v22.

发送消息

(fmtloggithub.com/apache/rocketmq-client-go/v2/producergithub.com/apache/rocketmq-client-go/v2/primitive

)func

producer.NewProducer(producer.WithNameServer([]string{localhost:9876}))if

err

p.SendSync(context.Background(),

msg)if

(contextfmtloggithub.com/apache/rocketmq-client-go/v2/consumergithub.com/apache/rocketmq-client-go/v2/primitive

)func

consumer.NewPushConsumer(consumer.WithNameServer([]string{localhost:9876}),consumer.WithGroupName(TestConsumerGroup),)if

err

定义消息处理函数c.RegisterMessageListener(consumer.MessageListenerFunc(func(ctx

context.Context,

consumer.go你应该能看到发送的消息被消费者接收到并输出。

注意事项

Group设置正确。

错误处理在生产环境中你应该实现更详细的错误处理和日志记录以便调试和监控。

语言中实践

是一个高性能、高可用的分布式消息中间件最初由阿里巴巴开源并在社区中逐渐发展和演进。

以下是

RocketMQ

的前身被阿里巴巴开发为一个内部消息中间件用于支持大规模的电商平台和业务系统。

2013年阿里巴巴开始开源

RocketMQ最初以阿里巴巴内部的需求和经验为基础开发。

早期版本主要用于阿里巴巴内部系统支持基本的消息队列功能。

2015年RocketMQ

提供了更加成熟的消息中间件功能满足了高吞吐量、低延迟和高可靠性的需求。

Apache

提供了生产环境所需的多种特性包括事务消息、消息过滤、消息顺序消费等。

SDK包括

2018年引入了新的特性和改进包括改进的消息存储机制、增强的高可用性和容错能力、以及更多的管理和监控工具。

RocketMQ

的高可用性、可扩展性和性能提供了更多的集群管理工具和性能监控功能。

RocketMQ

版本改进了消息顺序消费的可靠性和性能同时增强了对云平台的支持优化了与大数据平台的集成能力。

RocketMQ

版本增加了新的功能和改进包括更好的分布式事务支持、消息历史追踪功能和更高效的消息存储机制。

未来发展

继续关注云原生架构的演进、微服务架构的支持以及与其他大数据和流处理平台的集成。

未来的版本可能会引入更多对多租户和弹性伸缩的支持以及对新的消息通信协议的支持。

RocketMQ

从一个内部使用的消息中间件发展成为一个成熟的开源项目经过不断的改进和演进逐步成为

Apache

顶级项目并支持现代的云原生和微服务架构。

通过持续的开发和社区贡献RocketMQ

十一、一个秘密

https://gentlewok.blog.csdn.net/

锅总微信公众号