文档章节

rabbitmq——prefetch count

hncscwc
 hncscwc
发布于 2014/01/24 15:32
字数 1071
阅读 6341
收藏 13

消费者在开启acknowledge的情况下,对接收到的消息可以根据业务的需要异步对消息进行确认。

然而在实际使用过程中,由于消费者自身处理能力有限,从rabbitmq获取一定数量的消息后,希望rabbitmq不再将队列中的消息推送过来,当对消息处理完后(即对消息进行了ack,并且有能力处理更多的消息)再接收来自队列的消息。在这种场景下,我们可以通过设置basic.qos信令中的prefetch_count来达到这种效果。

先直观的看看设置了prefetch_count的效果,:

1) 对比测试:两个消费者都订阅同一队列,no_ack均设置为false即开启acknowledge机制,且均未设置prefetch_count,向队列发布5条消息

结果:不管消息是否被ack,rabbitmq会轮流向两个消费者投递消息,第一个消费者收到"1","3","5"三条消息, 第二个消费者收到"2","4"两条消息。

2)prefetch_count设置测试:两个消费者都订阅同一队列,开启acknowledge机制,第一个消费者prefetch_count设置为1,另一个消费者未设置prefetch_count,同样向队列发布5条消息

结果:rabbitmq向第一个消费者投递了一条消息后,消费者未对该消息进行ack,rabbitmq不会再向该消费者投递消息,剩下的四条消息均投递给了第二个消费者

看完效果后,再来看看rabbitmq里的一些实现。

1. rabbitmq对basic.qos信令的处理

首先,basic.qos是针对channel进行设置的,也就是说只有在channel建立之后才能发送basic.qos信令。

在rabbitmq的实现中,每个channel都对应会有一个rabbit_limiter进程,当收到basic.qos信令后,在rabbit_limiter进程中记录信令中prefetch_count的值,同时记录的还有该channel未ack的消息个数。

注:其实basic.qos里还有另外两个参数可进行设置(global和prefetch_size),但rabbitmq没有相应的实现。

2. 队列中的消息投递给消费者时的处理

当rabbitmq要将队列中的一条消息投递给消费者时,会遍历该队列上的消费者列表,选一个合适的消费者,然后将消息投递出去。其中挑选消费者的一个依据就是看消费者对应的channel上未ack的消息数是否达到设置的prefetch_count个数,如果未ack的消息数达到了prefetch_count的个数,则不符合要求。当挑选到合适的消费者后,中断后续的遍历。

rabbit_amqqueue_process.erl

deliver_msgs_to_consumers(_DeliverFun, true, State) ->
    {true, State};
deliver_msgs_to_consumers(DeliverFun, false,
                          State = #q{active_consumers =
                                     ActiveConsumers}) ->
    case priority_queue:out_p(ActiveConsumers) of
        {empty, _} ->
            {false, State};
        {{value, QEntry, Priority}, Tail} ->
            {Stop, State1} =
                deliver_msg_to_consumer(DeliverFun, QEntry,
                                        Priority,
                                        State#q{active_consumers =
                                                Tail}),
            %%如果处理结果为false,遍历下一个消费者
            deliver_msgs_to_consumers(DeliverFun, Stop, State1)
    end.

deliver_msg_to_consumer(DeliverFun, E = {ChPid, Consumer},
                        Priority, State) ->
    ...
    %%判断是否可以将消息投递给该消费者
    case rabbit_limiter:can_send(C#cr.limiter,
                                 Consumer#consumer.ack_required,
                                 Consumer#consumer.tag) of
        %%可以投递,再将该消费者放到队列的尾部
        {continue, Limiter} ->
            AC1 = priority_queue:in(E, Priority,
                                    State#q.active_consumers),
            %%将消息投递给消费者
            deliver_msg_to_consumer0(DeliverFun, Consumer,
                                     C#cr{limiter = Limiter},
                                     State#q{active_consumers = AC1})
    ...

rabbit_limiter.erl

handle_call({can_send, QPid, AckRequired}, _From,
            State = #lim{volume = Volume}) ->
    case prefetch_limit_reached(State) of
        %%未ack的消息数达到prefetch_count设置的个数
        true  -> {reply, false, limit_queue(QPid, State)};
        false -> {reply, true,
                  %%消息需要被ACK, volume加1
                  State#lim{volume = if AckRequired -> Volume + 1;
                                        true        -> Volume
                                     end}}
    end

prefetch_limit_reached(#lim{prefetch_count = Limit, 
                            volume = Volume}) ->
    Limit =/= 0 andalso Volume >= Limit.



3. 消费者对消息ack后的处理

当消费者对消息进行ack后,最终会修改该消费者对应channel中未ack的消息数,这样队列又可以将消息投递给该消费者。

rabbit_limiter.erl

handle_cast({ack, Count}, State = #lim{volume = Volume}) ->
    NewVolume = if Volume == 0 -> 0;
                   true        -> Volume - Count
                end,
    {noreply, maybe_notify(State, State#lim{volume = NewVolume})};



4. 扩展

在AMQP协议(0-9-1)中,有这么一段话

对于rabbitmq来说,最后一句话其实说的就是使用了acknowledge机制情况下,使用prefetch_count进行流量控制。另外在实际研究过程中发现还有channel.flow以及basic.credit(应该属于AMQP 1.0协议)可以进行一些控制,这里没有展开,有时间会研究下相应的机制以及可能使用的场景。

© 著作权归作者所有

共有 人打赏支持
上一篇: haproxy——配置
hncscwc
粉丝 67
博文 70
码字总数 76137
作品 0
杭州
程序员
私信 提问
加载中

评论(1)

chaun
chaun
79
RabbitMQ-C客户端使用说明

rabbitmq-c是一个用于C语言的,与AMQP server进行交互的client库,AMQP协议为版本0-9-1。rabbitmq-c与server进行交互前需要首先进行login操作,在操作后,可以根据AMQP协议规范,执行一系列操...

龙鸟
2012/09/20
0
0
python使用rabbitMQ介绍二(工作队列模式)

一模式介绍 第一章节的生产-消费者模式,是非常简单的模式,一发一收。在实际的应用中,消费者有的时候需要工作较长的时间,则需要增加消费者。 这时mq实现了一下几个功能: rabbitmq循环调度...

MyStitch
2018/07/30
0
0
RabbitMQ安装和使用详解(转载+实践)

1.解压缩.tar.gz文件: tar -zxvf 文件名称 mv 目标文件 目的地址 --移动文件 mv 原名称 新名称 --重命名 2.配置环境变量 # vi profile --编辑配置文件 在文档的最后添加: export PATH=$PAT...

hanzhankang
2014/02/20
0
0
RabbitMQ 入门【精+转】

rabbitmq可以用一本书取讲,这里只是介绍一些使用过程中,常用到的基本的知识点。 官方文档覆盖的内容,非常全面:http://www.rabbitmq.com/documentation.html 。 1. 介绍 RabbitMQ,即消息...

sunsky303
2018/05/01
0
0
(六)RabbitMQ消息队列-消息任务分发与消息ACK确认机制(PHP版)

在前面一章介绍了在PHP中如何使用RabbitMQ,至此入门的的部分就完成了,我们内心中一定还有很多疑问:如果多个消费者消费同一个队列怎么办?如果这几个消费者分任务的权重不同怎么办?怎么把...

Super_RD
2017/04/26
0
0

没有更多内容

加载失败,请刷新页面

加载更多

[git/tower]SSL certificate problem: Invalid certificate chain

fatal: unable to access 'https://xxx@130.51.23.250/baseline/mobile-framework/login-service.git/': SSL certificate problem: Invalid certificate chain 解决: git config --global ......

Danni3
17分钟前
0
0
ADI推出AD9528 JESD204B时钟和SYSREF发生器

根据ADI官网上对9361的介绍,其中还提到了与9361相配套的电源,时钟,LNA,PA等等功能部分需要的芯片,具体网页:https://www.analog.com/en/products/ad9361.html ADI近日宣布推出 AD9528 J...

whoisliang
30分钟前
1
0
Java springcloud B2B2C o2o多用户商城 springcloud架构-docker-feign配置(五)

简介 上一节我们讨论了怎么用feign声明式调用cloud的生产者,这节我们讨论一下feign配置,通过编写配置类,我们可以自定义feign的日志级别,日志扫描目录,可以通过feign调用服务在eureka上的...

sccspuercode
36分钟前
2
0
长连接的心跳及重连设计

前言 说道“心跳”这个词大家都不陌生,当然不是指男女之间的心跳,而是和长连接相关的。 顾名思义就是证明是否还活着的依据。 什么场景下需要心跳呢? 目前我们接触到的大多是一些基于长连接...

crossoverJie
36分钟前
7
0
OSChina 周三乱弹 —— 风扇写着先生请自爱

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @蚂蚁哈哈哈 :分享陈奕迅的单曲《落花流水》 《落花流水》- 陈奕迅 手机党少年们想听歌,请使劲儿戳(这里) @车谷 :我发现每天上班都好困 ...

小小编辑
今天
1K
15

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部