[rabbitmq-discuss] Multiple consumers

Mon Jul 9 16:13:38 BST 2007

Matthias,

> I gather that what you are trying to accomplish is to pull items off a
> queue and process them concurrently, correct?

Correct.

>
> The Java client libraries use one thread per connection, and the
> handleDelivery callback is executed in that. So in order to process
> messages concurrently you need to do one of the following:
>
> - use multiple connections
>
> - spawn a thread, perhaps from a pool, in handleDelivery
>
> - introduce some Java queuing / synchronisation to let handleDelivery
> pass messages to a pool of worker threads.
>
> There is some support for the last method in the next version of the
> client libraries, which introduces a QueuingConsumer class that places a
> deliveries into a Java queue from where they can be removed by worker
> threads. There are many possible alternative designs though. For
> example, you may want to use a bounded queue in order to get some flow
> control.

Understood.

>
> > I assume also that the same logic applies if you implement your
> > consumers using the native erlang API.
>
> Are you referring to the queue API that is used by the management
> service? In that case, yes, in essence you have similar design choices
> there on how to process the 'deliver' messages sent to the native queue
> process. Be aware though that with this API you are implementing a
> *queue*, not a *client*. So, for example, you end up blocking the
> channel process through which the message was sent until you return from
> the 'deliver' call.

Thanks for pointing this out. When you say that, do you mean the
defining a queue using the pseudo_queue method is implementing a queue
rather than a client, e.g.

Queue = rabbit_amqqueue:pseudo_queue(RealmName, QueueName, self())

Can I avoid this blocking by using the rabbit_amqqueue:basic_consume()
method instead?

>
> > BTW, what things do you have to consider when using the native API, if
> > you want to program your application in erlang. As far as I can tell,
> > it's pretty straightforward with the semantics being the same as the
> > java API except that you don't have the framing in the middle.
>
> That depends on what APIs you are using. Much of the AMQP semantics is
> concerned with notions of connections and channels, and much of the that
> is tied up with networking code in the rabbit_{reader,writer,channel}
> modules. Also, many aspects of the protocol, such as access control, are
> dealt with in rabbit_channel, since they are closely tied to the notion
> of a channel. So if you are bypassing all that and instead are directly
> calling into the modules for exchange, queue etc, then quite a big chunk
> of AMQP is cut out, though you are still left with the "essence" of
> routing and queuing.

Basically I have a remote consumers using the Java API and I have
implemented business services in Erlang, which has the *luxury* of
being able to call the API without having to frame methods and send
stuff over a wire. So I think it quite good to just be able to use the
asynchronous processing capabilities of the actual exchange.
Furthermore, if use Erlang on the server side, I don't need to worry
about concurrency issues and limitations that you would do in Java
(please correct me if I'm wrong).

>
> A should also point out that none of the Erlang APIs are official, and
> they *will* change.
>
> Several people have asked whether there is an Erlang client for AMQP,
> i.e. an Erlang API that implements the client portion of the
> protocol. There currently isn't, but it is definitely something we'd
> like to see developed. Contributions are welcome. Such a client would
> present a relatively stable and interoperable API for Erlang code
> wanting to use AMQP. It could also be designed in a way that allows the
> framing to be bypassed when talking to RabbitMQ.

Have you got any pointers as to how to go about this? I mean, would
you just define

a) The data structures as part of the interface spec (i.e. the
rabbit.hrl header file)
b) Methods to be exposed to client, e.g. openChannel0, openChannelN,
accessRequest, basicConsume, basicPublish, etc
c) Then implement these exported functions by sending the appropriate
messages to the appropriate internal servers?

Ben