|
|||||||||
PREV CLASS NEXT CLASS | FRAMES NO FRAMES | ||||||||
SUMMARY: NESTED | FIELD | CONSTR | METHOD | DETAIL: FIELD | CONSTR | METHOD |
java.lang.Object com.rabbitmq.client.DefaultConsumer com.rabbitmq.client.QueueingConsumer
public class QueueingConsumer
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().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
.
Nested Class Summary | |
---|---|
static class |
QueueingConsumer.Delivery
Encapsulates an arbitrary message - simple "bean" holder structure. |
Constructor Summary | |
---|---|
QueueingConsumer(Channel ch)
|
|
QueueingConsumer(Channel ch,
java.util.concurrent.BlockingQueue<QueueingConsumer.Delivery> q)
|
Method Summary | |
---|---|
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)
No-op implementation of Consumer.handleDelivery(java.lang.String, com.rabbitmq.client.Envelope, com.rabbitmq.client.AMQP.BasicProperties, byte[]) . |
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. |
Methods inherited from class com.rabbitmq.client.DefaultConsumer |
---|
getChannel, getConsumerTag, handleCancelOk, handleConsumeOk, handleRecoverOk |
Methods inherited from class java.lang.Object |
---|
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait |
Constructor Detail |
---|
public QueueingConsumer(Channel ch)
public QueueingConsumer(Channel ch, java.util.concurrent.BlockingQueue<QueueingConsumer.Delivery> q)
Method Detail |
---|
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 waiting
ShutdownSignalException
- if the connection is shut down while waiting
ConsumerCancelledException
- if this consumer is cancelled while waitingpublic QueueingConsumer.Delivery nextDelivery(long timeout) throws java.lang.InterruptedException, ShutdownSignalException, ConsumerCancelledException
timeout
- timeout in millisecond
java.lang.InterruptedException
- if an interrupt is received while waiting
ShutdownSignalException
- if the connection is shut down while waiting
ConsumerCancelledException
- if this consumer is cancelled while waiting
|
|||||||||
PREV CLASS NEXT CLASS | FRAMES NO FRAMES | ||||||||
SUMMARY: NESTED | FIELD | CONSTR | METHOD | DETAIL: FIELD | CONSTR | METHOD |