queue_speed_test.clj

(ns immutant-bench.queue-speed-test
  (:require [immutant.messaging :as m]))

;; Performance stats
;; 
;; Robs mac mini (4 cores)
;; -----------------------
;;
;; 10 producers, 5 consumers
;; 
;; 20k / 32s = 555 msgs/s
;;
;; Athlon(tm) 64 X2 Dual Core Processor 5600+ server
;; -------------------------------------------------
;;
;; 20k / ~44s = 450 msgs/s
;;
;; Ricks Imac (2.66 GHz Intel Core 2 Duo) 10 producers, 7 consumers
;; --------------------------------------
;;
;; 20k / 7s = 2857 msg/s

(def total-messages 20000)
(def q "some.queue")

(m/start q :durable false)

(defn listen [n-consumers p]
  (m/listen q #(when (= % "done")
                         (deliver p %)) :concurrency n-consumers))

(defn produce [producer-threads total-messages]
  (dotimes [n producer-threads]
    (future
      (immutant.messaging.core/with-connection {}
        (dotimes [n (/ total-messages producer-threads)]
          (m/publish q "a" :encoding :text :persistent false)))
      (m/publish q "done" :encoding :text :persistent false))))

(defn produce [pro-threads persist]
  (dotimes [n pro-threads]
    (future
      (immutant.messaging.core/with-connection {}
        (dotimes [n (/ total-messages pro-threads)]
          (m/publish q "a" :encoding :text :persistent persist)))
      (m/publish q "done" :encoding :text :persistent persist))))

(defn run 
  "Run test with a specified number of producer and consumer threads."
  ([n-producers n-consumers] (run n-producers n-consumers false))
  ([n-producers n-consumers persist]
     (let [p (promise)]
       (produce n-producers persist)
       (listen n-consumers p)
       @p)))


(comment  
  (time (run 12 6))
)

Slightly simplified from tobias's initial version, by using promises to sync instead of a polling loop.