basic.consume指的是channel在 某个队列上注册消费者,那在这个队列有消息来了之后,就会把消息转发到给此channel处理,如果 这个队列有多个消费者,则会采用轮转的方式将消息分发给消息者.
首先是rabbit_reader接收数据包后,解析组装出其中的method,channel方法交给channel处理.具体过程见http://www.cnblogs.com/haoqingchuan/p/4354692.html
channel进程处理basic.consume的方法.先从状态中查看是否已经存在此tag(以channel为域,不同的consumer_tag标识了不同的消费者,每个channel的内的consumer tag必须是唯一的).如果没有查找到则正常,如果未对队列名字命名,则会产生一个uuid来作为队列名.
handle_method(#'basic.consume'{queue = QueueNameBin,
consumer_tag = ConsumerTag,
no_local = _, % FIXME: implement
no_ack = NoAck,
exclusive = ExclusiveConsume,
nowait = NoWait,
arguments = Args},
_, State = #ch{consumer_prefetch = ConsumerPrefetch,
consumer_mapping = ConsumerMapping}) ->
case dict:find(ConsumerTag, ConsumerMapping) of
error ->
QueueName = qbin_to_resource(QueueNameBin, State),
check_read_permitted(QueueName, State),
ActualConsumerTag =
case ConsumerTag of
<<>> -> rabbit_guid:binary(rabbit_guid:gen_secure(),
"amq.ctag");
Other -> Other
end,
case basic_consume(
QueueName, NoAck, ConsumerPrefetch, ActualConsumerTag,
ExclusiveConsume, Args, NoWait, State) of
{ok, State1} ->
{noreply, State1};
{error, exclusive_consume_unavailable} ->
rabbit_misc:protocol_error(
access_refused, "~s in exclusive use",
[rabbit_misc:rs(QueueName)])
end;
{ok, _} ->
%% Attempted reuse of consumer tag.
rabbit_misc:protocol_error(
not_allowed, "attempt to reuse consumer tag '~s'", [ConsumerTag])
end;
basic_consume(QueueName, NoAck, ConsumerPrefetch, ActualConsumerTag,
ExclusiveConsume, Args, NoWait,
State = #ch{conn_pid = ConnPid,
limiter = Limiter,
consumer_mapping = ConsumerMapping}) ->
case rabbit_amqqueue:with_exclusive_access_or_die(
QueueName, ConnPid,
fun (Q) ->
{rabbit_amqqueue:basic_consume(
Q, NoAck, self(),
rabbit_limiter:pid(Limiter),
rabbit_limiter:is_active(Limiter),
ConsumerPrefetch, ActualConsumerTag,
ExclusiveConsume, Args,
ok_msg(NoWait, #'basic.consume_ok'{
consumer_tag = ActualConsumerTag})),
Q}
end) of
{ok, Q = #amqqueue{pid = QPid, name = QName}} ->
CM1 = dict:store(
ActualConsumerTag,
{Q, {NoAck, ConsumerPrefetch, ExclusiveConsume, Args}},
ConsumerMapping),
State1 = monitor_delivering_queue(
NoAck, QPid, QName,
State#ch{consumer_mapping = CM1}),
{ok, case NoWait of
true -> consumer_monitor(ActualConsumerTag, State1);
false -> State1
end};
{{error, exclusive_consume_unavailable} = E, _Q} ->
E
end.
rabbit_amqqueue.erl
rabbitmq_channel进程向rabbitmq_amqp_process进程发送消息来完成增加消费者的动作
basic_consume(#amqqueue{pid = QPid, name = QName}, NoAck, ChPid, LimiterPid,
LimiterActive, ConsumerPrefetchCount, ConsumerTag,
ExclusiveConsume, Args, OkMsg) ->
ok = check_consume_arguments(QName, Args),
delegate:call(QPid, {basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
ConsumerPrefetchCount, ConsumerTag, ExclusiveConsume,
Args, OkMsg}).
rabbit_amqqueue_process.erl
增加consumer，并更新到state中。
handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
PrefetchCount, ConsumerTag, ExclusiveConsume, Args, OkMsg},
_From, State = #q{consumers = Consumers,
exclusive_consumer = Holder}) ->
case check_exclusive_access(Holder, ExclusiveConsume, State) of
in_use -> reply({error, exclusive_consume_unavailable}, State);
ok -> Consumers1 = rabbit_queue_consumers:add(
ChPid, ConsumerTag, NoAck,
LimiterPid, LimiterActive,
PrefetchCount, Args, is_empty(State),
Consumers),
ExclusiveConsumer =
if ExclusiveConsume -> {ChPid, ConsumerTag};
true -> Holder
end,
State1 = State#q{consumers = Consumers1,
has_had_consumers = true,
exclusive_consumer = ExclusiveConsumer},
ok = maybe_send_reply(ChPid, OkMsg),
emit_consumer_created(ChPid, ConsumerTag, ExclusiveConsume,
not NoAck, qname(State1),
PrefetchCount, Args, none),
notify_decorators(State1),
reply(ok, run_message_queue(State1))
end;
rabbit_queue_consumers.erl
更新进程字典,并为队列增加新消费者.
add(ChPid, CTag, NoAck, LimiterPid, LimiterActive, Prefetch, Args, IsEmpty,
State = #state{consumers = Consumers,
use = CUInfo}) ->
C = #cr{consumer_count = Count,
limiter = Limiter} = ch_record(ChPid, LimiterPid),
Limiter1 = case LimiterActive of
true -> rabbit_limiter:activate(Limiter);
false -> Limiter
end,
C1 = C#cr{consumer_count = Count + 1, limiter = Limiter1},
update_ch_record(
case parse_credit_args(Prefetch, Args) of
{0, auto} -> C1;
{_Credit, auto} when NoAck -> C1;
{Credit, Mode} -> credit_and_drain(
C1, CTag, Credit, Mode, IsEmpty)
end),
Consumer = #consumer{tag = CTag,
ack_required = not NoAck,
prefetch = Prefetch,
args = Args},
State#state{consumers = add_consumer({ChPid, Consumer}, Consumers),
use = update_use(CUInfo, active)}.
%%将consumer加入consumers列表里面,也就是后面分发消息的时候会从这个列表里将消息取出
in(X, 0, { queue, [_] = In, [], 1}) ->
{queue, [X], In, 2};
in(X, 0, {queue, In, Out, Len}) when is_list(In), is_list(Out) ->
{queue, [X|In], Out, Len + 1};
