diff --git a/docs/guides/messaging.md b/docs/guides/messaging.md index c91e816e..e38f4425 100644 --- a/docs/guides/messaging.md +++ b/docs/guides/messaging.md @@ -148,6 +148,16 @@ dereference: The responder is just a fancy listener, and can be deregistered the same way as a listener. +## Concurrency + +Listeners can have multiple threads invoking their handler as messages +are consumed. You control how many with the `:concurrency` option +provided by [[listen]] and [[respond]]. By default, it is set to 1 for +topics and the number of available processors for queues, but for +IO-bound handlers, you may see better performance as you increase the +number. It very much depends on what your handler is doing and how +many messages it needs to process concurrently. + ## Durable Topic Subscriptions Typically, messages published to a topic are only delivered to diff --git a/messaging/src/immutant/messaging.clj b/messaging/src/immutant/messaging.clj index a7d68e2b..52e00bc5 100644 --- a/messaging/src/immutant/messaging.clj +++ b/messaging/src/immutant/messaging.clj @@ -227,7 +227,7 @@ The following options are supported [default]: - * :concurrency - the number of threads handling messages [1] + * :concurrency - the number of threads handling messages [1 for topics, #cores for queues] * :selector - A JMS (SQL 92) expression matching message metadata/properties [nil] * :decode? - if true, the decoded message body is passed to `f`. Otherwise, the base message object is passed [true] diff --git a/messaging/src/immutant/messaging/pipeline.clj b/messaging/src/immutant/messaging/pipeline.clj index 6741584f..1de5c0b0 100644 --- a/messaging/src/immutant/messaging/pipeline.clj +++ b/messaging/src/immutant/messaging/pipeline.clj @@ -258,7 +258,7 @@ * :name a name to use for the step [the current index of the fn] * :concurrency the number of threads to use, overriding the pipeline - setting [1] + setting [#cores] * :decode? if false, the raw message object will be passed to this step [true] * :error-handler an error handler function that can override the @@ -301,7 +301,7 @@ * :concurrency the number of threads to use for *each* step. Can be overridden on a per-step basis - see the 'step' - function. [1] + function. [#cores] * :error-handler a function that will be called when any step raises an exception. It will be passed the exception and the argument to the step. Without an error-handler, diff --git a/project.clj b/project.clj index defbe02e..8b167528 100644 --- a/project.clj +++ b/project.clj @@ -91,7 +91,7 @@ clj-http "1.0.1" ;; org.projectodd.wunderboss "0.4.1" - org.projectodd.wunderboss "1.x.incremental.220" + org.projectodd.wunderboss "1.x.incremental.224" ;; org.projectodd.wunderboss "0.5.1-SNAPSHOT" org.immutant :version