『互联网架构』软件架构-rocketmq之特性和运维
小编:管理员 174阅读 2022.07.29
(一)Producer详解rocketmq的producer 和 Consumer的特性比较了解。 源码:https://github.com/limingios/netFuture/tree/master/jms 中的rocketmq-test项目
- 普通消息
涉及到的类 org.apache.rocketmq.client.impl.CommunicationMode org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendDefaultImpl
- 定时消息
消息发到 Broker 后,不能立刻被 Consumer 消费,要到特定的时间点或者等待特定的时间后才能被消费。 固定精度: 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
- 顺序消息
涉及到的类 org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendSelectImpl 在分布式队列实现排序,这个需求怎么做? 优先级:订单号20180206(queu1 )订单号20180207(queue2) 排序+优先级(分片)
- 事物消息
- Pull&push模式
DefaultMQPullConsumer DefaultMQPushConsumer
- 消费模型
org.apache.rocketmq.common.protocol.heartbeat.MessageModel#BROADCASTING org.apache.rocketmq.common.protocol.heartbeat.MessageModel#CLUSTERING
- 消费选择
org.apache.rocketmq.common.consumer.ConsumeFromWhere#CONSUMEFROMLASTOFFSET org.apache.rocketmq.common.consumer.ConsumeFromWhere#CONSUMEFROMFIRSTOFFSET org.apache.rocketmq.common.consumer.ConsumeFromWhere#CONSUMEFROMTIMESTAMP
- 重试策略
Consumer 消费消息失败后,要提供一种重试机制,令消息再消费一次。Consumer 消费消息失败
- 消息重复幂
RocketMQ无法避免消息重复,所以如果业务对消费重复非常敏感,务必要在业务层面去重
- 消息回溯
回溯消费是指Consumer已经消费成功的消息,由于业务上需求需要重新消费,要支持此功能。 RocketMQ支持按照时间回溯消费,时间维度精确到毫秒,可以向前回溯,也可以向后回溯。 使用命令:sh mqadmin resetOffsetByTime -g xx -s yy -t tt
- Producer|consumer Group:
- 可以通过运维工具(上次说过的rocketmq-console)查询这个组下有多少Producer实例。
- 可通过运维工具查询这个组下的消费进度,多少个Consumer实例。
- 事务消息,如果Producer意外宕机,Broker会主动回调Producer Group中的任意一台机器确认事务状态。
- 集群模式,一个Consumer Group下的多个Consumer均摊消费消息;广播模式,group无意义。
- 作用
- 每个Broker启动的时候会向Namesrv发送注册请求,Namesrv接收Broker的请求注册路由信息,NameServer保存活跃的broker列表,包括Master和Slave。
- 用来保存所有topic和该topic所有队列的列表。
- NameServer用来保存所有broker的Filter列表。
- 接收client(Producer和Consumer)的请求根据某个topic获取所有到broker的路由信息。 总结:维护路由信息、维护broker数据。
- 启动过程
org.apache.rocketmq.namesrv.NamesrvStartup org.apache.rocketmq.namesrv.NamesrvController#initialize org.apache.rocketmq.namesrv.kvconfig.KVConfigManager#load org.apache.rocketmq.remoting.netty.NettyRemotingServer(初始化远程服务、线程池服务) 初始化过程执行
- 读取配置文件
- 调用初始化方法#108
- 注册broker处理接受消费和消费消息等功能
- 定时去扫描不活跃的broker
- 作用
消息接受、存储(恢复)、与个节点之间通讯。
- 启动过程>org.apache.rocketmq.broker.BrokerStartup#createBrokerController#210
org.apache.rocketmq.common.ConfigManager#load#加载文件(store/config)内容 org.apache.rocketmq.store.DefaultMessageStore#load加载磁盘文件内容 org.apache.rocketmq.remoting.netty.NettyRemotingServer初始化通信层 线程池 org.apache.rocketmq.broker.BrokerController#registerProcessor 注册远程服务
- DefaultMessageStore
持久核心处理类
- FlushConsumeQueueService >逻辑队列刷盘服务
- CleanCommitLogService 清理物理文件服务,定期清理72小时之前的物理文件。
- CleanConsumeQueueService 定期清理在逻辑队列中的物理偏移量小于commitlog中的最小物理偏移量的数据
- HAService 用于commitlog数据的主
- ScheduleMessageService 用于监控延迟消息,并到期后执行
- TransactionStateService 用于事务消息状态文件
初始化过程执行
- 初始化>createBrokerController>initialize
- 加载topic、消费进度、订阅关系
- 加载本地消息messageStore.load
- 远程通讯NettyRemotingServer、加载处理消息类。5
- 初始化发送线程池sendMessageExecutor、拉取消息线程池(pullMessageExecutor)、管理Broker线程池(adminBrokerExecutor)、客户端管理线程池(clientManageExecutor)。
- 注册事件处理器,包括发送消息事件处理器(SendMessageProcessor)、拉取消息事件处理器、查询消息事件处理器(QueryMessageProcessor,包括客户端的心跳事件、注销事件、获取消费者列表事件、更新更新和查询消费进度consumerOffset)、客户端管理事件处理器、结束事务处理器(EndTransactionProcessor)、默认事件处理器(AdminBrokerProcessor)。
相关推荐
- 【RocketMQ系列】RocketMQ集群,RocketMQ-on-DLedger集群 本文RocketMQ系列第四篇,主要介绍RocketMQ集群及如何部署自动容灾切换的 RocketMQ-on-DLedger Group。RocketMQ集群搭建ROcketMQ集群搭建有以下几种方案:「单Master模式」「多Master模式」「多Master多Slave模式-异步复制」「多Master多Slave模式-同步双写」其…
- 3DMAX提示和技巧 本主题标识使用 Civil View 的一些重要提示和技巧。常规使用屏幕分辨率至少为 1280x1024 的 Civil View。低于此分辨率时,一些面板将占用过多屏幕空间。 将视口设置为线框显示以达到最佳性能。 要尽可能简化用户界面,请在单个视口中工作并关闭 3ds Max 命令面…