QueueingConsumer

Class Overview

Convenience class: an implementation of Consumer with straightforward blocking semantics. The general pattern for using QueueingConsumer is as follows:

 // Create connection and channel.
 ConnectionFactory factory = new ConnectionFactory();
 Connection conn = factory.newConnection();
 Channel ch1 = conn.createChannel();

 // Declare a queue and bind it to an exchange.
 String queueName = ch1.queueDeclare().AMQP.Queue.DeclareOk#getQueue getQueue();
 ch1.queueBind(queueName, exchangeName, queueName);

 // Create the QueueingConsumer and have it consume from the queue
 QueueingConsumer consumer = new QueueingConsumer(ch1);
 ch1.basicConsume(queueName, false, consumer);

 // Process deliveries
 while (/* some condition * /) {
     QueueingConsumer.Delivery delivery = consumer.nextDelivery();
     // process delivery
     ch1.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
 }
 

For a more complete example, see LogTail in the test/src/com/rabbitmq/examples directory of the source distribution.

deprecated QueueingConsumer was introduced to allow applications to overcome a limitation in the way Connection managed threads and consumer dispatching. When QueueingConsumer was introduced, callbacks to Consumers were made on the Connection's thread. This had two main drawbacks. Firstly, the Consumer could stall the processing of all Channels on the Connection. Secondly, if a Consumer made a recursive synchronous call into its Channel the client would deadlock.

QueueingConsumer provided client code with an easy way to obviate this problem by queueing incoming messages and processing them on a separate, application-managed thread.

The threading behaviour of Connection and Channel has been changed so that each Channel uses a distinct thread for dispatching to Consumers. This prevents Consumers on one Channel holding up Consumers on another and it also prevents recursive calls from deadlocking the client.

As such, it is now safe to implement Consumer directly or to extend DefaultConsumer.

Summary

[Expand] Inherited Methods
From class com.rabbitmq.client.DefaultConsumer Channel getChannel() Retrieve the channel. String getConsumerTag() Retrieve the consumer tag. void handleCancel(String consumerTag) No-op implementation of handleCancel(String) void handleCancelOk(String consumerTag) No-op implementation of handleCancelOk(String). void handleConsumeOk(String consumerTag) Stores the most recently passed-in consumerTag - semantically, there should be only one. void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) No-op implementation of handleDelivery(String, Envelope, AMQP.BasicProperties, byte[]). void handleRecoverOk(String consumerTag) No-op implementation of handleRecoverOk(String). void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) No-op implementation of handleShutdownSignal(String, ShutdownSignalException).
From class java.lang.Object Object clone() boolean equals(Object arg0) void finalize() final Class<?> getClass() int hashCode() final void notify() final void notifyAll() String toString() final void wait() final void wait(long arg0, int arg1) final void wait(long arg0)
From interface com.rabbitmq.client.Consumer abstract void handleCancel(String consumerTag) Called when the consumer is cancelled for reasons other than by a call to basicCancel(String). abstract void handleCancelOk(String consumerTag) Called when the consumer is cancelled by a call to basicCancel(String). abstract void handleConsumeOk(String consumerTag) Called when the consumer is registered by a call to any of the basicConsume(String, boolean, Consumer) methods. abstract void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) Called when a basic.deliver is received for this consumer. abstract void handleRecoverOk(String consumerTag) Called when a basic.recover-ok is received in reply to a basic.recover. abstract void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) Called when either the channel or the underlying connection has been shut down.