Spring Integration provides several different Message Channel implementations. Each is briefly described in the sections below.
The PublishSubscribeChannel
implementation broadcasts any Message
sent to it to all of its subscribed consumers. This is most often used for sending
Event Messages whose primary role is notification as opposed to
Document Messages which are generally intended to be processed by
a single consumer. Note that the PublishSubscribeChannel
is
intended for sending only. Since it broadcasts to its subscribers directly when its
send(Message)
method is invoked, consumers cannot poll for
Messages (it does not implement PollableChannel
and
therefore has no receive()
method). Instead, any subscriber
must be a MessageConsumer
itself, and the subscriber's
send(Message)
method will be invoked in turn.
The QueueChannel
implementation wraps a queue. Unlike, the
PublishSubscribeChannel
, the QueueChannel
has point-to-point
semantics. In other words, even if the channel has multiple consumers, only one of them should receive any
Message sent to that channel. It provides a default no-argument constructor (providing an essentially unbounded
capacity of Integer.MAX_VALUE
) as well as a constructor that accepts the queue capacity:
public QueueChannel(int capacity)
A channel that has not reached its capacity limit will store messages in its internal queue, and the
send()
method will return immediately even if no receiver is ready to handle the
message. If the queue has reached capacity, then the sender will block until room is available. Likewise, a
receive call will return immediately if a message is available on the queue, but if the queue is empty, then
a receive call may block until either a message is available or the timeout elapses. In either case, it is
possible to force an immediate return regardless of the queue's state by passing a timeout value of 0.
Note however, that calling the no-arg versions of send()
and
receive()
will block indefinitely.
Whereas the QueueChannel
enforces first-in/first-out (FIFO) ordering, the
PriorityChannel
is an alternative implementation that allows for messages to be ordered
within the channel based upon a priority. By default the priority is determined by the
'priority
' header within each message. However, for custom priority determination
logic, a comparator of type Comparator<Message<?>>
can be provided to the
PriorityChannel
's constructor.
The RendezvousChannel
enables a "direct-handoff" scenario where a sender will block
until another party invokes the channel's receive()
method or vice-versa. Internally,
this implementation is quite similar to the QueueChannel
except that it uses a
SynchronousQueue
(a zero-capacity implementation of
BlockingQueue
). This works well in situations where the sender and receiver are
operating in different threads but simply dropping the message in a queue asynchronously is too dangerous. For
example, the sender's thread could roll back a transaction if the send operation times out, whereas with a
QueueChannel
, the message would have been stored to the internal queue and potentially
never received.
The RendezvousChannel
is also useful for implementing request-reply
operations. The sender can create a temporary, anonymous instance of RendezvousChannel
which it then sets as the 'replyChannel' header when building a Message. After sending that Message, the sender
can immediately call receive (optionally providing a timeout value) in order to block while waiting for a reply
Message.
The DirectChannel
has point-to-point semantics, but otherwise is more similar to the
PublishSubscribeChannel
than any of the queue-based channel implementations described
above. It implements the SubscribableChannel
interface instead of the
PollableChannel
interface, so it dispatches Messages directly to a subscriber.
As a point-to-point channel, however, it differs from the PublishSubscribeChannel
in
that it will only send each Message to a single subscribed
MessageConsumer
. Its primary purpose is to enable a single thread to perform the
operations on "both sides" of the channel. For example, if a consumer is subscribed to a
DirectChannel
, then sending a Message to that channel will trigger invocation of that
consumer's onMessage(Message)
method directly in the sender's
thread. The key motivation for providing a channel implementation with this behavior is to support
transactions that must span across the channel while still benefiting from the abstraction and loose coupling
that the channel provides. If the send call is invoked within the scope of a transaction, then the outcome of
the consumer's invocation (e.g. updating a database record) can play a role in determining the ultimate result
of that transaction (commit or rollback).
Note | |
---|---|
Since the DirectChannel is the simplest option and does not add any additional
overhead that would be required for scheduling and managing the threads of a poller, it is the default
channel type within Spring Integration. The general idea is to define the channels for an application and
then to consider which of those needs to provide buffering to throttle input, and to modify those to be
queue-based PollableChannels . Likewise, if a channel needs to broadcast
messages, it should not be a DirectChannel but rather a
PublishSubscribeChannel . Below you will see how these can be configured.
|
The final channel implementation type is ThreadLocalChannel
. This channel also delegates
to a queue internally, but the queue is bound to the current thread. That way the thread that sends to the
channel will later be able to receive those same Messages, but no other thread would be able to access them.
While probably the least common type of channel, this is useful for situations where
DirectChannels
are being used to enforce a single thread of operation but any reply
Messages should be sent to a "terminal" channel. If that terminal channel is a
ThreadLocalChannel
, the original sending thread can collect its replies from it.