轻松搞定RocketMQ入门

  • 时间:
  • 浏览:1
  • 来源:彩神幸运飞艇_神彩幸运飞艇官方

(3)前端停留系统进程向用户返回成功。

RocketMQ不止也能直接推送消息,在消费端注册监听器进行监听,还也能由消费端决定自己去拉取数据





开始英文英文的没办法 细看PullResult对象,以为拉取到的结果没办法 MessageExt对象还跑到群后边问别人,犯2了

2.根据slotValue(slot对应位置的值)查找到索引项列表的最后一项(倒序排列,slotValue经常指向最新的有另一个 索引项)

在有 RAID 卡, SAS 500000 转磁盘测试顺序写文件,数率也能达到 500M 每秒左右,而线上的网卡一般都为千兆网卡,写磁盘数率明显快于数据网络入口数率,没办法 与否也能做到写完 内存就向用户返回,由后台系统进程刷盘呢?

可能业务流程支持批量法子消费,则也能很大程度上的提高吞吐量,也能通过设置Consumer的consumerMessageBatchMaxSize参数,默认是1,即一次消费根小参数

(2)每个消息在业务层面的唯一标识码,要设置到 keys 字段,方便将来定位消息丢失问題。服务器会为每个消息创建索引(哈希索引),应用也能通过 topic,key 来查询这条消息内容,以及消息被谁消费。可能是哈希索引,请务必保证 key 尽可能唯一,可是也能正确处理潜在的哈希冲突。

(1)写嘴笨 是顺序写,有已经 读却变成了随机读

(2).使业务层面的情形机去重

(1). 可能磁盘数率大于网卡数率,没办法 刷盘的进度肯定也能跟上消息的写入数率。

也能保证严格的消息顺序

实时的消息订阅机制

优点:也能利用DMA法子,消耗CPU资源少,大块文件传输数率高,无内存安全新问題

(2)可能Consume Queue存储数量极少,有已经 顺序读,在pagecache的与读取情形下,Consume Queue的读性能与内存几乎经常,即使堆积情形下。统统也能认为Consume Queue完整性不需要阻碍读性能

1.下载系统进程

在有Slave情形下,Master一旦发现Consumer访问堆积在磁盘的数据时,回想Consumer下达有另一个 重定向指令,令Consumer从Slave拉取数据,可是正常的发消息与正常的消费不需要可能堆积受影响,可能系统将堆积场景与否堆积场景分割在了有另一个 不同的节点正确处理。这里会产生有另一个 问題,Slave会不需要写性能下降,答案与否定的。可能Slave的消息写入只追求吞吐量,不追求实时性,假使 整体的吞吐量高就行了,而Slave每次前要从Master拉取一批数据,如1M,你这名批量顺序写入法子使堆积情形,整体吞吐量影响相对较小,可是写入RT会变长。

一阵一阵要注意 静态变量offsetTable的作用,拉取的是按照从offset(理解为下标)位置开始英文拉取,拉取N条,offsetTable记录下次拉取的offset位置。

为你这名过滤要没办法 做?

优点:即使频繁调用,使用小文件块传输,数率也很高

a )使用mmap+write法子

(2)读根小消息,会先读Consume Queue,再读Commit Log,增加了开销

RocketMQ无法做到消息重复,统统可能业务对消息重复非常敏感,务必要在业务层面去重,有以下某些法子:

缺点:小块文件数率低于mmap法子,也上能 是BIO法子传输,也能 使用NIO

2.过滤过程中不需要访问Commit Log 数据,也能保证堆积情形下也能高效过滤

4.Hash冲突,寻找key的slot位置时大慨执行了两次散列函数,一次key的hash,一次key的hash取值模,有已经 这里处于两次冲突的情形;第两种,key的hash值不同但模数相同,此时查询的事先 会在比较第一次key的hash值(每个索引项保存了key的hash值),过滤掉hash值让你等的情形。第二种,hash值相等key让你等,出于性能的考虑冲突的检测倒进客户端正确处理(key的原始值是存储在消息文件中的,正确处理对数据文件的解析),客户端比较一次消息体的key与否相同

(4)Consumer与NameServer集群中的其蕴含另一个 节点(随机选则)建立长连接,定期从NameServer取Topic路由信息,并向提供Topic服务的Master、Slaver建立长连接,且定时向Master、Slaver发送心跳。Consumer即可从Master订阅消息,也也能从Slave订阅消息,订阅规则由Broker配置决定

我是在虚拟机中的CentOS6.5中进行部署。

如:

(1)队列轻量化,单个队列数据量非常少

3.Producer发送消息,由异步系统进程刷盘,消息从pagecache刷入磁盘

可能调用 send 同步法子发送失败,则尝试将消息存储到 db,由后台系统进程定时重试,保证消息一定到达 Broker。

1 消息的堆积容量、依赖磁盘大小

2.2同步刷盘:

2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解压到适当的目录如/opt/目录

3.1按照MessageId查询消息



MsgId总共16个字节,蕴含消息储存主机地址,消息Commit Log Offset。从MsgId中解挥发Broker的地址和Commit Log 偏移地址,有已经 按照存储格式所在位置消息buffer解析成有另一个 完整性消息

b)使用sendfile法子

推荐阅读:JAVA-ACE-架构师系列- RocketMQ

有另一个 RPC 调用,通常是可是有另一个 过程

(2)服务器正确处理该请求



1.Producer发送消息,消息从socket进入java 堆

b) 可能干净页欠缺,此时写入 PAGECACHE 会被阻塞,系统尝试刷盘要素数据,大慨每次尝试 32 个 PAGE,来找出更多干净 PAGE。

综上,内存溢出的情形不需要经常经常出现

也能查看sameple中的quickstart源码 1.Consumer 消息消费者



2.Producer消息生产者



3.首先运行Consumer系统进程,经常在运行情形接收服务器端推送过来的消息



4.再次运行Producer系统进程,生成消息并发送到Broker,Producer的日志冲没哟,有已经 也能看完Broker推送到Consumer的根小消息

每个方案前要优缺点,他的缺点是:

3 正常消费的Consumer与否会受影响、无Slave情形,会受一定影响、有Slave情形,不受影响

上述 db 重试法子为你这名没办法 集成到 MQ 客户端內部做,可是要求应用自己去完成,基于以下几点考虑:

a) 写入消息到 PAGECACHE 时,可能内存欠缺,则尝试丢弃干净的 PAGE,腾出内存供新消息使用,策略是 LRU 法子。

(5)对于消息不可丢失应用,务必要有消息重发机制

(1)零拷贝原理:Consumer消费消息过程,使用了零拷贝,零拷贝包括一下2中法子,RocketMQ使用第两种法子,因小块数据传输的要求效果比sendfile法子好

SLAVE_NOT_AVAILABLE:消息发送成功,有已经 此时 slave 不可用,消息可能进入服务器队列,也上能 此时服务器宕机,消息才会丢失。对于精确发送顺序消息的应用,可能顺序消息的局限性,可能会涉及到主备自动切换问題,统统可能sendresult 中的 status 字段不等于 SEND_OK,就应该尝试重试。对于某些应用,则没办法 必要可是

(2) 可能发送失败,则轮转到下有另一个 Broker

(1)MQ 的客户端设计为无情形模式,方便任意的水平扩展,且对机器资源的消耗仅仅是 cpu、内存、网络

处于消息堆积时,可能消费数率经常跟不上发送数率,也能选则丢弃不重要的消息



如以上代码所示,当某个队列的消息数堆积到 5000000 条以上,则尝试丢弃要素或完整性消息,可是就也能快速追上发送消息的数率

2.Consumer收到过滤消息后,同样也要执行在broker端的操作,有已经 比对的是真实的Message Tag字符串,而前要hashcode

这条消息的消费过程与 DB 交互了 4 次,可能按照每次 5ms 计算,没办法 总共耗时 20ms,假设业务计算耗时 5ms,没办法 总过耗时 25ms,可也能把 4 次 DB 交互优化为 2 次,没办法 总耗时就也能优化到 15ms,也可是说总体性能提高了 40%。

(2)数据存储特征

(2)Broker部署相对复杂,Broker氛围Master与Slave,有另一个 Master也能对应多个Slaver,有已经 有另一个 Slaver也上能 对应有另一个 Master,Master与Slaver的对应关系通过指定相同的BrokerName,不同的BrokerId来定义,BrokerId为0表示Master,非0表示Slaver。Master也能部署多个。每个Broker与NameServer集群中的所有节点建立长连接,定时注册Topic信息到所有的NameServer

SEND_OK:消息发送成功

(3)消息发送成功可能失败,要打印消息日志,务必要打印 sendresult 和 key 字段

可是做的好处:

文章来源:https://segmentfault.com/a/1190000015951993

亿级消息堆积能力

1.单机支持1W以上的持久化队



(1)所有数据单独储存到commit Log ,完整性顺序写,随机读

1.Message Tag存储hashcode,是为了在Consume Queue定长法子存储,节约空间

(2)刷盘系统进程刷盘后,唤醒前端停留系统进程,可能是一批系统进程。

FLUSH_DISK_TIMEOUT:消息发送成功,有已经 服务器刷盘超时,消息可能进入服务器队列,也上能 此时服务器宕机,消息才会丢失

缺点:也能 很好的利用DMA法子,会比sendfile多消耗CPU资源,内存安全性控制复杂,前要正确处理JVM Crash问題

(1).将消息的唯一键,能与否MsgId,也能与否消息内容中的唯一标识字段,这名订单ID,消费事先 判断与否在DB或Tair(全局KV存储)中处于,可能不处于则插入,并消费,有已经 跳过。(实践过程要考虑原子性问題,判断与否处于也能尝试插入,可能报主键冲突,则插入失败,直接跳过) msgid一定是全局唯一的标识符,有已经 可能会处于同样的消息有有另一个 不同的msgid的情形(有多种导致 ),你这名情形可能会使业务上重复,建议最好使用消息体中的唯一标识字段去重

(1)随机读,尽可能让读命中pagecache,减少IO操作,统统内存越大越好。可能系统中堆积的消息很多,读数据要访问硬盘会不需要可能随机读导致 系统性能急剧下降,答案与否定的。

(3)Producer 所在机器的可靠性较低,一般为虚拟机,不适合存储重要数据。 综上,建议重试过程交由应用来控制。

7.同5

4 访问堆积在磁盘的消息时,吞吐量有多大、与访问的并发有关,最终会降到50000左右

5.Consumer拉消息(异常消费),消息直接从pagecache转入socket

4.Consumer拉消息(正常消费),消息直接从pagecache(数据在物理内存)转入socket,到达Consumer,不经过java堆。你这名消费场景最多,线上96G物理内存,按照1K消息算,也能物理缓存1亿条消息

5.存储,为了节省空间索引项中存储的时间是时间差值(存储时间——开始英文时间,开始英文时间存储在索引文件头中),整个索引文件是定长的,特征也是固定的

b)随机访问Commit Log 磁盘数据,系统IO调度算法设置为NOOP法子,会在一定程度上将完整性的随机读变成顺序跳跃法子,而顺序跳跃法子读较完整性的随机读性能高5倍

高效的订阅者水平扩展能力

(2)对磁盘的访问串行话,正确处理磁盘竞争,不需要可能队列增加导致 IOWait增高

主要导致 是线下压测时间过短,线上运行一段时间后,cache 命中率下降,没办法 RT 就会增加。建议在线下压测时,要测试足够长时间,尽可能模拟线上环境,压测过程中,数据的分布也一阵一阵要,数据不同,可能 cache 的命中率也会完整性不同

统统有另一个 RPC 的耗时时间是上述有另一个 步骤的总和,而某些场景要求耗时非常短,有已经 对可靠性要求暂且高,这名日志分类分类整理类应用,此类应用也能采用 oneway 形式调用,oneway 形式只发送请求不停留应答,而发送请求在客户端实现层面仅仅是有另一个 os 系统调用的开销,即将数据写入客户端的 socket 缓冲区,此过程耗时通常在微秒级。

(1)客户端发送请求到服务器

举这名下,某条消息的消费过程如下

1、根据消息从 DB 查询数据 1

2、根据消息从 DB 查询数据2

3、复杂的业务计算

4、向 DB 插入数据3

5、向 DB 插入数据 4

(2)可能 MQ 客户端內部集成有另一个 KV 存储模块,没办法 数据也上能 同步落盘也能较可靠,而同步落盘两种性能开销较大,统统通常会采用异步落盘,又可能应用关闭过程不受 MQ 运维人员控制,可能经常会处于 kill -9 可是暴力法子关闭,造成数据没办法 及时落盘而丢失

个别应用可能会遇到你这名情形:在线下压测消费过程中,db 表现非常好,每次 RT 都很短,有已经 上线运行一段时间,RT 就会变长,消费吞吐量直线下降

6.Consumer拉消息(异常消费),可能socket访问了虚拟内存,产生缺页中断,此前要产生磁盘IO,从磁盘Load消息到pagecache,有已经 直接从socket发出去

(2)对最终用户展现的队列实际只储存消息在Commit Log 的位置信息,有已经 串行法子刷盘

2 发消息的吞吐量大小受影响程度、无Slave情形,会受一定影响、有Slave情形,不受影响

FLUSH_SLAVE_TIMEOUT:消息发送成功,有已经 服务器同步到 Slave 时超时,消息可能进入服务器队列,也上能 此时服务器宕机,消息才会丢失

提供充裕的消息拉取模式



(1)NameServer是有另一个 几乎无情形的节点,可集群部署,节点之间无任何信息同步

(3)服务器向客户端返发表声明答

(3) 你这名法子的总耗时时间不超过 sendMsgTimeout 设置的值,默认 10s统统,可能两种向 broker 发送消息产生超时异常,就不需要再做重试

(1) 至多重试 3 次

8.同6

以上缺点怎么客服:

rocketmq中的所有消息前要持久化的,先写入系统pagecache,有已经 刷盘,也能保证内存与磁盘前要一份数据,访问时,也能直接从内存读取

2.1异步刷盘

3.遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的32条记录)

(1)写入 PAGECACHE 后,系统进程停留,通知刷盘系统进程刷盘。

Producer 的 send 法子两种支持內部重试,重试逻辑如下:

(3)Producer与NameServer集群中的其蕴含另一个 节点(随机选则)建立长连接,定期从NameServer取Topic路由信息,并向提供Topic服务的Master建立长连接,且定时向Master发送心跳。Produce完整性无情形,可集群部署

a)访问pagecache时,即使只访问1K的消息,系统也会提前预读出更多的数据,在下次读时就可能命中pagecache

3.2按照Message Key查询消息



1.根据查询的key的hashcode%slotNum得到具体的槽位置 (slotNum是有另一个 索引文件后边蕴含的最大槽目数目,这名图中所示slotNum=5000W)

对于 Mysql 等 DB,可能部署在磁盘,没办法 与 DB 进行交互,可能数据没办法 命中 cache,每次交互的 RT 会直线上升, 可能采用 SSD,则 RT 上升趋势要明显好于磁盘。

同步刷盘与异步刷盘的唯一区别是异步刷盘写完 PAGECACHE 直接返回,而同步刷盘前要停留刷盘完成才返回,同步刷盘流程如下:

3.即使处于hash冲突,也也能在Consumer端进行修正,保证万无一失

3.启动RocketMQ:进入rocketmq/bin 目录 执行



4.启动Broker,设置对应的NameServer

2.Producer发送消息,消息从java堆进入pagecache,物理内存

(2). 万一可能此时系统压力过大,可能堆积消息,除了写入 IO,还有读取 IO,万一经常经常出现磁盘读取落后情形,会不需要导致 系统内存溢出,答案与否定的,导致 如下:

(4)send 消息法子,假使 不抛异常,就代表发送成功。有已经 发送成功会有多个情形,在 sendResult 里定义

(3)要保证Commit Log 与 Consume Queue完整性的一致,增加了编程的复杂度

(3)Commit Log中存储了所有的元信息,蕴含消息体,这名于MySQl、Oracle的redolog,统统假使 有Commit Log处于, Consume Queue即使丢失数据,仍也能恢复出来

(1) 有另一个 应用尽可能用有另一个 Topic,消息子类型用 tags 来标识,tags 也能由应用自由设置。也上能 发送消息设置了tags,消费方在订阅消息时,也也能利用 tags 在 broker 做消息过滤。

2.刷盘策略

RocketMQ的消息过滤法子有别于某些的消息后边件,是在订阅时,再做过滤,先来看下Consume Queue存储特征



1.在Broker端进行Message Tag比较,先遍历Consume Queue,可能存储的Message Tag与订阅的Message Tag不符合,则跳过,继续比对下有另一个 ,符合则传输给Consumer。注意Message Tag是字符串形式,Consume Queue中存储的是其对应的hashcode,比对时也是比对hashcode