问题一 RocketMq 消费流程
获取topic 对应的消费客户端和所有的broker 下的Que队列 然后根据一定的算法分配本客户端要拉取的QueID
分配算法
1 环行平均分配算法，平均然后轮流分配
q1 q2 q3 q4 q5 q6 q7 q8 3个消费队列 c1 c2 c3
c1:q1 q4 q7
c2:q2 q5 q8
c3:q3 q6
2 平均分配
q1 q2 q3 q4 q5 q6 q7 q8 3个消费队列 c1 c2 c3
c1: q1 q2 q3
c2: q4 q5 q6
c3: q7 q8
3机房优先平均分配
优先分配同机房的消息队列，然后平均
4自定义算法指定客户端消费某些队列算法。
5 一致性hash算法
算法原理 ：
步骤1 构造clientid的hash环,TreeMap 为集合，hash值作为key，节点为clientid
步骤2 计算que的hash 节点，获取> 本hash值得最近得一个clientid节点。
1 多线程并行处理，不同得队列并行拉取数据，消息并行消费
默认一次拉取32消息，节省网络带宽
2 并发消费，不会因为中间得一个消息有问题，就停顿卡进度
一个个消息消费，根据结果 反馈broker，有问题仍然 反馈给服务端进入重试队列，进行下一次得消费重试，不会因为有问题导致消费进度得卡顿
3 即使有一个消息有问题死循环，有超时检测机制
对待处理得消息队列进行超时判断，超过了时间没处理完毕，发送回broker ，并从内存中移除。
答案
答案： 无论批次消费，还是一个个消费，坐标以小得为准。
目的： 防止后面得消费完，前面得消息还没消费完，服务宕机了，导致消息丢失。
源码：
存在得问题：为了确保消息不丢失，服务器重启得时候，会导致重复消费。
顺序消费的目的:客户端消费消息是按照，消息进入的顺序，并发消息，offset大的消息，有可能先于offset小的消息先行消费完。
1消费逻辑上，顺序消费，必须锁定broker 对应的消息队列，防止重新负载的时候分配给其他client
2顺序消费，一次拉取32条消息，如果中间有一条消息卡滞，消费失败，后面的消息挂起，这条消息重试16次，
如果还是失败，就会发回服务端。跳过此次消费。
问题七
顺序消费上了几把锁，为什么要上锁
1 负载均衡的时候，队列发生变化
目的：负载变化，要求broker给队列上锁，变更期间不允许分配给其他client
2数据拉取后，对队列数据的加锁，保持队列的顺序性消费
对集合上锁。
1 消息重试机制，消费失败的消息会重新发回broker端，
2 broker收到ack 响应才认为消费成功，否则不认为是成功
3客户端拉取一批消息，即使后面的先于前面的消费完，即使broker宕机，也只更新低的offset 确保消息不丢失
1消费失败会发回重试。根据重试的次数， 发往不同等级的重试队列
定时取出消息发往原来的topic 和que
达到最大失败次数放入死信队列
消息堆积有几种原因
消息堆积监控
1判断是否存在消息堆积场景
1producer发送消息的速率监控
2producer发送消息的maxOffset与consumer消费消息的currOffset的差异值与给定的消息堆积数值告警值对比，如果差异 值大于数据告警值，则存在消息堆积，否则不存在消息堆积。
3consumer消费消息的速率监控
通过扩容能解决问题的现象
1 突然流量激增，导致堆积。
2 Broker消息堆积，比如Broker的性能瓶颈，Broker同步策略导致消息堆积等
3 consumer本身已经拉取消息的堆积。consumer消息拉取超过一定量之后会暂停消息拉取，一方面是消费者本身消费能力的现在，另一方面是由于消费端过多的消息容易造成GC频繁。
扩容还解决不了的问题，还存在挤压现象，就要考虑broker 或client本身的故障
这种情况基本上是可以确定是RocketMQ本身的故障照成的，比如Broker故障，比如Broker的GC频率过高导致消息推送，copy性能降低，集群内部网络故障，等等。此时主要是监控RocketMQ服务器性能，或消费逻辑有问题
感谢以下作者辛苦的劳作参考
>消息的发布是指某个生产者向某个topic发送消息；消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息，进而从该topic消费数据。
消息发送后进入同步等待状态，可以保证消息投递一定到达。可靠的同步消息传输可用于重要的通知消息，SMS通知，SMS营销系统等广泛的场景中。
异步传输通常用于响应时间敏感的业务场景。想要快速发送消息，又不想丢失的时候可以使用异步消息。
只发送消息，不等待服务器响应，只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。单向消息适用于需要适度可靠性的场景，例如日志收集。
可以多条消息打包一起发送，减少网络传输次数提高效率。
producersend(Collection c) 方法可以接收一个集合实现批量发送。
批量发送消息的复杂度只会在发送大批量并且你可能不确定它是否超过了大小限制（1MiB）时增加。在这种情况下，你最好将列表拆分：
消息消费模式由消费者来决定，可以由消费者设置MessageModel来决定消息消费模式。
消息模式默认为集群消费模式。
RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。RocketMQ支持两种消息过滤方式，一种是在Broker端进行过滤，另一种是在Consumer端进行过滤。
在Broker中，按照Consumer的要求做过滤，优点是减少了对于Consumer无用消息的网络传输。
缺点是增加了Broker的负担，实现相对复杂。
这种过滤方式可由应用完全自定义实现，但是缺点是很多无用的消息要传输到Consumer端。
在大多数情况下，TAG是一个简单而有用的设计，其可以用来选择你想要的消息。例如：
消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。
SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。下面是一个例子：
RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。
在 brokerconf 中添加配置：
启动 broker 加载指定配置文件：
随后在集群配置中可以看到:
发送消息时，你可以通过putUserProperty来设置消息的属性：
可以通过MessageSelectorbySql来筛选消息不存在任何直接的冲突，不过它们也不能完全兼容。Yncsrver主要是分布式服务器管理系统，它能够让用户通过多台服务器节点来提供服务。而rocketmg是一款用于服务器集群管理的工具，它能够帮助用户将多台服务器或集群分配到多个不同的应用中，并能够将任务调度、延迟任务、安全等功能统一管理。总而言之，它们可以共存且相互补充，但是要看具体的业务需求而定。
说起零拷贝之前，先来了解下服务器中文件数据通过网络传输到客户端的流程。作为应用服务器，其中会有很多从磁盘中读取数据，然后应用程序对加载到内存中的数据进行处理，然后通过网卡发送给客户端，传统数据处理通过以下两个函数实现：
在这个过程中，数据流转的大致过程如下：
可以见到，在这个过程中发生了2次cpu copy和2次DMA copy，以及发生了数次cpu状态切换。 这个 *** 作对于应用服务器来说很频繁，因此带来的开销也是非常大。
因此所谓的零拷贝就是，让其中的2次cpu拷贝省略掉，因为这两次cpu拷贝的数据其实已经在内存中，没有必要再让cpu参与进来进行数据的拷贝，浪费cpu。在大量文件读写的时候，这个优化带来的收益还是比较可观的。
零拷贝的实现方式有两种：
mmap通过虚拟内存映射，让多个虚拟地址指向同一个物理内存地址，用户空间的虚拟地址和内核空间的虚拟地址指向同一个物理内存地址，这样用户空间和内核空间共享同一个内存数据。这样DMA引擎从磁盘上加载的数据不需要在内核空间和用户空间进行复制，减少了一次cpu拷贝。
sendfile通过系统调用，并且规定了in_fd文件描述符必须是可以mmap的，sendfile只能将文件数据发送到socket中，sendfile减少了一次cpu状态的切换
无论是mmap结合write方式还是sendfile方式都只是减少了一次cpu拷贝，而后DMA引擎还具有了收集功能，可以在内核缓存区发送到socket缓冲区的时候避免掉cpu复制，只是将缓冲区地址和数据长度发送给socket缓冲区，然后DMA引擎通过收集功能直接读取收集数据发送到网卡中。这里依赖DMA引擎的收集功能省略掉了最后一次cpu拷贝，到此才是真正的零拷贝。
所谓的零拷贝就是避免数据在内核空间缓存区和用户空间缓缓冲区之间的复制，避免掉2次cpu复制，释放cpu。
在RocketMq中采用的是mmap()结合write()方式来实现零拷贝。
在java中还可以通过FileChanneltransferTo()来实现数据从文件描述符传输到socket中，它的底层是通过sendfile系统调用来实现。
rocketmq是阿里巴巴开源的mq，目前在github拥有13+k的star。rocketmq是众多mq实现中，较少使用java实现的，因此对于java技术栈的人来说，拿rocketmq的源码作为切入点，理解mq的实现原理是非常合适的。本文会从四大部分（namesrv、broker、producer、consumer）讲解rocketmq源码，之间的关系可见rocketmq架构图。
namesrv是类似zk的命名服务端，broker向它发起注册、producer与consumer向他拉取topic的队列。为什么不用现有的比如zk等中间件呢？应该是因为解耦：功能比较简单不需要引入外界中间件，避免引入新的复杂度，控制权在自己手上，简单即是美。
来到NamesrvStartup，里面定义里main方法，启动时会对NamesrvController进行创建，然后调用start方法对其进行启动，而启动过程是先initialize再start，打扫干净再迎客。
先看initialize，此方法主要对kvConfigManager进行加载、初始化远程服务器并注册处理器、初始化两个定时任务（扫描不活跃的broker、打印周期日志）。
再看start，主要是启动远程服务器，对本地端口进行绑定。
那broker怎么么向它发起注册？producer与consumer怎么向他拉取topic的队列？由上文可知，在初始化方法已经向远程服务器注册了处理器DefaultRequestProcessor，当请求进来时会流向processRequest方法。
我们挑选一些核心的请求处理进行解析：
1REGISTER_BROKER
大于等于V3_0_11版本由registerBrokerWithFilterServer处理，主要是调用RouteInfoManager的registerBroker进行注册。
来到registerBroker，主要是对五个map存放broker相关信息，
clusterAddrTable存放的是clusterName与brokerName的对应关系。
brokerAddrTable存放的是brokerName与brokerAddr的对应关系。
brokerLiveTable存放的是brokerAddr与brokerLiveInfo的对应关系。
filterServerTable存放的是brokerAddr与filterServerList的对应关系。
topicQueueTable存放的是topicName与queueDataList的对应关系。
2GET_ROUTEINFO_BY_TOPIC
调用getRouteInfoByTopic，继续通过routeInfoManager的pickupTopicRouteData方法获取topicRouteData。
来到pickupTopicRouteData，实际上就是通过topicQueueTable获取到队列信息，然后根据brokerName从brokerAddrTable获取到brokerData，最后根据brokerAddr获取到filterServerList，进行返回。
来到broker的BrokerStartup，先创建brokerController在调用start启动broker。
先看createBrokerController，里面对配置进行处理，然后创建BrokerController并进行initialize。
initialize方法有点长，主要是对manager（topicConfigManager、consumerOffsetManager等）的加载，messageStore的初始化，线程池的生成（请求处理、心跳、落盘、日志等），生成远程服务器并且注册处理器。
而start会调用controller的start进行处理，其实就是调用各个start方法和向namesrv注册。
还记得注册的处理器吗？我们看下主要的处理器源码。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）