前言
MQ现在在互联网公司算是一个必不可少的中间件了，我们都知道MQ可以用来流量削峰，异步解耦，但是总觉得只停留在最基本的使用上。正好最近看了一遍《RocketMQ实战与原理解析》这本书，讲的内容不见得有多高深、多全面，但是看完了回忆一遍，好多东西感觉只有一个模糊的印象，还是分章记下来可靠一点。
快速入门
主要介绍了MQ在实际中的应用场景
应用解耦
非关键链路服务挂掉后，MQ可以缓存消息，等服务再启动再去完成，不影响主链路的运行。
流量削峰
可以使用普通性能的服务器加消息队列来应对高峰期请求，节约开支
消息分发
数据的产生方只需要把各自的数据写人一个消息队列即可，数据使用方根据各自需求订阅感兴趣的数据，不同数据团队所订阅的数据可以重复也可以不重复，互不干扰，也 不必和数据产生方关联。
最终一致性、动态扩容
生产环境下的配置和使用
角色——邮局
- Producer——发信者
- Consumer——收信者
- Broker——负责暂存、传输的邮局
- NameServer——负责协调各个地方邮局的管理机构
image
常用配置参数
- namesrvAddr=192.168.100.131:9876;192.168.100.132:9876
NamerServer的地址，可以是多个
- brokerClusterName=DefaultCluster
Cluster 的地址，如果集群机器数比较多，可以分成多个Cluster，每个Cluster供一个业务群使用
- brokerName=broker-b
Broker 的名称， Master 和 Slave 通过使用相同的 Broker 名称来表明相互关系，以说明某个 Slave 是哪个 Master 的 Slave
- brokerid=0
一个 Master Barker可以有多个 Slave, 0表示 Master，大于 0表示不同 Slave 的 ID
- deleteWhen=04
与 fileReservedTime参数呼应，表明在几点做消息删除动作，默认值04表示凌晨4点
- fileReservedTime=48
在磁盘上保存消息的时长，单位是小时，自动删除超时的消息
- brokerRole=SYNC_MASTER
brokerRole 有 3 种: SYNC_MASTER、ASYNC_MASTER、SLAVE。 关键 词 SYNC 和 ASYNC 表示 Master 和 Slave 之间同步消息的机制， SYNC 的意思 是当 Slave 和 Master 消息同步完成后，再返回发送成功的状态
- flushDiskType=ASYNC_FLUSH
flushDiskType表示刷盘策略，分为SYNC_FLUSH和ASYNC_FLUSH两种，分别代表同步刷盘和异步刷盘。 同步刷盘情况下，消息真正写人磁盘后再返回成功状态;异步刷盘情况下，消息写人 page_cache 后就返回成功状态
- listenPort=10911
Broker监听的端口号，如果一台机器上启动了多个 Broker， 则要设置不同的端口号，避免冲突
- storePathRootDir=/home/rocketmq/store-b
存储消息以及一些配置信息的根目录
常用管理命令
- 创建/修改 Topic
- 删除 Topic
- 创建/修改订阅组
- 删除订阅组
- 更新 Broker配置
- 更新 Topic 的读写权限
- 查询 Topic 的路由信息
- 查看 Topic 列表信息
- 查看 Topic 统计信息
- 根据时间查询消息
- 根据消息 ID 查询消息
- 查看集群消息
图形界面管理
rocketmq-console
用适合的方式发送和接收消息
消费者
- DefaultMQPushConsumer
- 由系统控制读取操作，收到消息后自动调用传人的 处理方法来处理
- Clustering：由 Broker 端存储和控制 Offset 的值， Broadcasting：Offset存到本地
- DefaultMQPul!Consumer
- 读取操作中的大部分功 能由使用者自主控制
- 自己处理Offset
生产者
- DefaultMQProducer
- 发往指定的Message Queue，使用MessageQueueSelector
消息模式
-
Clustering
同一个 ConsumerGroup(GroupName相同) 里的每个 Consumer 只消费所订阅消息的一部分内容，同一个 ConsumerGroup 里所有的 Consumer消费的内容合起来才是所订阅 Topic 内容的整体，从而达到负载均衡的目的 -
Broadcasting
同一个 ConsumerGroup里的每个 Consumer都能消费到所订阅 Topic 的全部消息，也就是一个消息会被多次分发，被多个 Consumer消费
分布式消息队列的协调者
NameServer功能
- NameServer本身是无状态的，也就是说 NameServer 中的 Broker、Topic 等状态信息不会持久存储，都是由各个角色定时上报并存储到内存中的(NameServer支持配置参数的持久化，一般用不到)
- 代表集群状态的五个变量，org.apache.rocketmq.namesrv.routeinfo.RoutelnfoManager
- private final HashMap<String/* topic */, List<QueueData>> topicQueueTable topicQueueTable
topic映射broker
- private final HashMap<String/* BrokerName */, BrokerData> BrokerAddrTable
brokerName映射broker
- private final HashMap<String/* ClusterName */, Set<String/* BrokerName */>> ClusterAddrTable
- private final HashMap<String/* BrokerAddr */, BrokerLivelnfo> BrokerLiveTable
BrokerAddr映射broker信息
- private final HashMap<String/* BrokerAddr */, List<String>/* Filter Server*/> filterServerTable
BrokerAddr映射过滤服务器
底层通信机制
image
image
- 核心命令类——RemotingCommand
消息队列的核心机制
- 使用mmap实现“零拷贝”
消息存储结构
image
- 核心——ConsumeQueue和CommitLog
CommitLog顺序写，随机读，好处：
- 顺序写，提高写入效率
- 随机读，利用操作系统的 pagecache 机制，加速后续的读取速度
- 为了保证完全的顺序写，需要 ConsumeQueue 这个中间结构 ，因为 ConsumeQueue 里只存偏移量信息，所以尺寸是有限的，在实际情况中，大部分的 ConsumeQueue 能够被全部读人内存，所以这个中间结构的操作速度很快，可以认为是内存读取的速度 。 此外为了保证 CommitLog 和 ConsumeQueue 的一 致性， CommitLog 里存储了 ConsumeQueues、 Message key、 Tag 等所有信息， 即使 ConsumeQueue 丢失，也可以通过 commitLog 完全恢复出来
高可用性机制
- 消费端：Master和Slave，自动切换
- 发送端：把 Topic 的多个 Message Queue创建在多个 Broker组上
同步刷盘和异步刷盘
image
同步复制和异步复制
同步复制方式是 等 Master 和 Slave 均写成功后才反馈给客户端写成功状态;异步复制方式是只要 Master 写成功即可反馈给 客户端写成功状态。
可靠性优先的使用场景
顺序消息
- 全局顺序——基本不用
- 部分顺序
- 要保证部分消息有序，需要发送端和消费端配合处理。在发送端，要做到把同一业务 ID 的消息发送到同一个 Message Queue ;在消费过程中，要做到从同一个 Message Queue 读取的消息不被并发处理，这样才能达到部分有序。
- 发送端使用MessageQueueSelector，消费端使用MessageListenerOrderly
- 在 MessageListenerOrderly 的实现中，为每个 Consumer Queue 加个锁，消费每个消息前，需要先获得这个消息对应的 Consumer Queue 所对应的锁，这样保证了同一时间，同一个 Consumer Queue 的消息不被并发消费，但
不同 Consumer Queue 的消息可以并发处理 。
消息重复问题
- 幂等
- 维护已消费消息的记录
动态增减机器(NameServer、Broker)，故障对消息影响，消息优先级
吞吐量优先的使用场景
在 Broker 端进行消息过滤
- 通过 Tag 进行过滤
- 用 SQL 表达式的方式进行过滤
- Filter Server方式过滤
提高 Consumer 处理能力
- 提高消费并行度
- 以批量方式进行消费
- 检测延时情况，跳过非重要消息
Consumer 的负载均衡
提高 Producer 的发送速度
- Oneway 方式只发送请求不等待应答，即将数据写人客户端的Socket缓冲区就返回，不等待对方返回结果
- 增加 Producer 的并发量
主从同步机制
- 同步属性信息
- 同步消息体——CommitLog同步不是经过 netty command 的方式， 而是直接进行 TCP 连接，这样效率更高。
网友评论