public class QueueingConsumer extends DefaultConsumer
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().getQueue
(); ch1.queueBind
(queueName, exchangeName, queueName); // Create the QueueingConsumer and have it consume from the queue QueueingConsumer consumer = newQueueingConsumer
(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.
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
and QueueingConsumer
is a lot less relevant.
Modifier and Type | Class and Description |
---|---|
static class |
QueueingConsumer.Delivery
Encapsulates an arbitrary message - simple "bean" holder structure.
|
Constructor and Description |
---|
QueueingConsumer(Channel ch) |
QueueingConsumer(Channel ch,
java.util.concurrent.BlockingQueue<QueueingConsumer.Delivery> q) |
Modifier and Type | Method and Description |
---|---|
void |
handleCancel(java.lang.String consumerTag)
No-op implementation of
Consumer.handleCancel(String) |
void |
handleDelivery(java.lang.String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
|
void |
handleShutdownSignal(java.lang.String consumerTag,
ShutdownSignalException sig)
No-op implementation of
Consumer.handleShutdownSignal(java.lang.String, com.rabbitmq.client.ShutdownSignalException) . |
QueueingConsumer.Delivery |
nextDelivery()
Main application-side API: wait for the next message delivery and return it.
|
QueueingConsumer.Delivery |
nextDelivery(long timeout)
Main application-side API: wait for the next message delivery and return it.
|
getChannel, getConsumerTag, handleCancelOk, handleConsumeOk, handleRecoverOk
public QueueingConsumer(Channel ch)
public QueueingConsumer(Channel ch, java.util.concurrent.BlockingQueue<QueueingConsumer.Delivery> q)
public void handleShutdownSignal(java.lang.String consumerTag, ShutdownSignalException sig)
DefaultConsumer
Consumer.handleShutdownSignal(java.lang.String, com.rabbitmq.client.ShutdownSignalException)
.handleShutdownSignal
in interface Consumer
handleShutdownSignal
in class DefaultConsumer
consumerTag
- the consumer tag associated with the consumersig
- a ShutdownSignalException
indicating the reason for the shut downpublic void handleCancel(java.lang.String consumerTag) throws java.io.IOException
DefaultConsumer
Consumer.handleCancel(String)
handleCancel
in interface Consumer
handleCancel
in class DefaultConsumer
consumerTag
- the defined consumer tag (client- or server-generated)java.io.IOException
public void handleDelivery(java.lang.String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws java.io.IOException
DefaultConsumer
Consumer.handleDelivery(java.lang.String, com.rabbitmq.client.Envelope, com.rabbitmq.client.AMQP.BasicProperties, byte[])
.handleDelivery
in interface Consumer
handleDelivery
in class DefaultConsumer
consumerTag
- the consumer tag associated with the consumerenvelope
- packaging data for the messageproperties
- content header data for the messagebody
- the message body (opaque, client-specific byte array)java.io.IOException
- if the consumer encounters an I/O error while processing the messageEnvelope
public QueueingConsumer.Delivery nextDelivery() throws java.lang.InterruptedException, ShutdownSignalException, ConsumerCancelledException
java.lang.InterruptedException
- if an interrupt is received while waitingShutdownSignalException
- if the connection is shut down while waitingConsumerCancelledException
- if this consumer is cancelled while waitingpublic QueueingConsumer.Delivery nextDelivery(long timeout) throws java.lang.InterruptedException, ShutdownSignalException, ConsumerCancelledException
timeout
- timeout in millisecondjava.lang.InterruptedException
- if an interrupt is received while waitingShutdownSignalException
- if the connection is shut down while waitingConsumerCancelledException
- if this consumer is cancelled while waiting