RocketMQ知识整理与总结

一、架构

 

　　RocketMQ的master broker与master broker没有任何消息通信，nameserver之间也一样没有消息通讯

　　MQ历史 　　　　

　　　　由数据结构队列发展而来 　　

　　MQ使用场景

　　 　　　   异步处理
　　　　　　解耦
　　　　　　削峰填谷
　　　　　　数据同步

二、队列

　　rocketMQ一个主题（topic）包含多个队列

　　

 

三、使用

　　生产

　　　　同步（sync）

　　　　　　默认重试2次总共3次

　　　　　　默认等待超时时间为3s 　　　　

　　　　异步（async）

　　　　　　总共重试2次

　　　　单向（oneway）

　　　　Message

　　　　　　topic：主题名称

　　　　　　tag：消息TAG，用于消息过滤对消息的总体分类，好比 topic为物流跟踪轨迹 ，轨迹包含 揽收 出库 入库 派送 签收，能够分别给这些相同topic不一样类型的数据打标签分类解析处理

　　　　　　keys：Message索引键，多个用空格隔开，RocketMQ能够根据这些key快速检索到消息对消息关键字的提取方便查询，好比一条消息某个关键字是 运单号，以后咱们可使用这个运单号做为关键字进行查询
　　　　　　waitStoreMsgOK：消息发送时是否等消息存储完成后再返

　　　　　　delayTimeLevel：消息延迟级别，用于定时消息或消息重

　　　　　　User property：自定义消息属性

　　　　批量发送

　　　　　　单批次消息不能超过maxMessageSize大小（默认4M）

　　　　　　客户端instance：若是instance为默认值DEFAULT的话，RocketMQ会自动将instance设置为IP+进程ID（建议不要设置，默认生成就好），默认最大4M

　　　　　　钩子方法：能够执行先后通知

　　　　事物消息

　　　　　　1分钟回查一次，默认5次

　　　　　　事物消息单独一篇 　　

　　消费

  　　　　　 批量消费总数为32，broker设置

　　　　　　若是消息消费次数超过maxReconsumeTimes还未成功，则将该消息转移到一个失败队列，等待被删除

　　　　　　消息消费超时时间，默认为15分钟

　　　　　　消息最大重试次数，默认为16次

　　　　　　consumeConcurrentlyMaxSpan，并发消息消费时处理队列最大跨度，默认2000，表示若是消息处理队列中偏移量最大的消息与偏移量最小的消息的跨度超过2000则延迟50毫秒后再拉取消息

　　　　　　pullInterval=0，推模式下拉取任务间隔时间，默认一次拉取任务完成继续拉取

　　　　　　consumeMessageBatchMaxSize：消息并发消费时一次消费消息条数，通俗点说就是每次传入MessageListtener#consumeMessage中的消息条数

　　　　　　RocketMQ消息重试是以消费组为单位，而不是主题，消息重试主题名为％RETRY％+消费组名。消费者在启动的时候会自动订阅该主题，参与该主题的消息队列负载

　　　　　　同一个消息队列只会分配给一个消费者，故若是消费者个数大于消息队列数量，则有些消费者没法消费消息。

　　　　　　若是延迟级别大于0，则会将消息的主题设置为SCHEDULE_TOPIC_XXXX

　　　　　　transactionId 事物ID会本身生成

　　　　　　ConsumeFromWhere

　　　　　　　　CONSUME_FROM_FIRST_OFFSET：从头开始消费

　　　　　　　　ONSUME_FROM_TIMESTAMP：从消费者启动的时间戳对应的消费进度开始消费

　　　　　　　　CONSUME_FROM_LAST_OFFSET：从队列最新偏移量开始消费

　　　　　　　　CONSUME_SUCCESS：消费成功

　　　　　　　　RECONSUME_LATER：延迟消费，放弃本批次消息消费 相似于continue，若是有重试次数没有达到最大上限会再次消费

　　　　消息消费模式

　　　　　　集群模式：默认模式，主题下的同一条消息只容许被其中一个消费者消费

　　　　　　　　　　   消费进度存储在服务端

　　　　　　广播模式：主题下的同一条消息将被集群内的全部消费者消费一次

　　　　　　　　　　　消费进度存储在消费者本地

　　　　消息传输模式

　　　　　　拉取消息模式：消费端主动发起拉消息请求

　　　　　　　　　　　　长轮询模式使得消息拉取能实现准实时

　　　　　　　　　　　　从服务器拉取消息->放入内存队列->提交消息处处理线程池

　　　　并发消费

　　 　　　　推送消息模式：RocketMQ消息推模式的实现基于拉模式

　　　　　　RocketMQ并无真正实现推模式，而是消费者主动向消息服务器拉取消息，RocketMQ推模式是循环向消息服务端发送消息拉取请求

　　　　　　单独线程池拉取消息，而后调用监听api接口

　　　　　　单独线程池拉取->内存队列->消息处理线程池处理->移除客户端内存队列消息并更新进度

　　　　顺序消息

　　　　　　消费过程 消息队列负载->消息拉取->消息消费->消息消费进度存储。

　　　　　　支持局部顺序消息消费，也就是保证同一个消息队列上的消息顺序消费

　　　　　　若是要实现某一主题的全局顺序消息消费，能够将该主题的队列数设置为1，牺牲高性能和可用性

　　　　　　顺序消息在建立消息队列拉取任务时须要在Broker服务器锁定该消息队列。

　　　　　　MAX_TIME_CONSUME_CONTINUOUSLY：每次消费任务最大持续时间，默认为60s，切换线程

　　　　　　顺序消息消费的并发度为消息队列。也就是一个消息消费队列同一时刻只会被一个消费线程池中一个线程消费。

　　　　　　达到重试次数上限，转移到死信队列，继续后续消息的消费

　　　　定时消息

　　　　　　消息发送以后并不当即被消费者消费，而是要等到特定的时间以后才能被消费

　　　　　　不支持任意时间精度定时发送，只支持配置级别的时间默认为"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"，delayLevel=1表示延迟1s，delayLevel=2表示延迟5s，依次类推。

　　　　　　SCHEDULE_TOPIC_XXXX定时消息主题

　　　　消息过滤

　　　　　　tag

　　　　　　　　tag服务端只是验证了TAG的hashcode，客户端再次对消息进行tag值对比过滤

　　　　　　sql(SQL92表达式)

　　　　　　　　（官方示例有bug）表达式没有想象的好用，建议你们接收到消息本身判断筛选

　　　　　　类过滤：定制过滤消息

　　　　　　　　消息过滤服务器（不讲解）

　　　　　　　　consumer->filterserver->broker

　　　　单向消息（One-way）

　　　　　　其实就是UDP协议的实现

　　　　　　TCP协议是可靠消息传输协议，请求消息都会有相应和校验，在会话层和传输层解决应答

四、填坑方案 　　

　　若是有大量消息积压

　　　　增长消费者数量

　　若是有大量消息积压而且立刻就到了自动清理的时间

　　　　从新消费导流到新的topic，增大新topic的队列数量

五、bug 　　

　　netty epoll 4.4.0以前版本没有实现

六、问题 　　

　　为何某条消息报异常会阻塞整个队列消费 　　　

　　　　ProcessQueue中队列最大偏移量与最小偏离量的间距，不能超过consumeConcurrentlyMaxSpan，不然触发流控。

　　　　这里主要的考量是担忧一条消息堵塞，消息进度没法向前推动，可能形成大量消息重复消费

七、初始化客户端注解 　　

　　使用@PostConstruct 由JSR-250提供，在构造函数执行完以后执行，等价于xml配置文件中bean的initMethod

　　若是同一个jvm中同时注入生产者和消费者使用bean注解会有异常抛出

八、客户端支持的驱动程序 　　

　　Java

　　Go

　　.net

　　Php

　　c++

　　Nodejs