Потребитель производителя с квалификациями

Вот мое мнение. Я решил использовать только структуры данных Clojure, чтобы увидеть, как это сработает. Обратите внимание, что было бы совершенно обычным и идиоматическим решением взять очередь блокировки из набора инструментов Java и использовать ее здесь; я думаю, код будет легко адаптировать. Обновление: Я действительно адаптировал его к java.util.concurrent.LinkedBlockingQueue , см. Ниже.

clojure.lang.PersistentQueue

Вызов (pro-con) , чтобы начать тестовый прогон; затем просмотрите содержимое вывода output , чтобы узнать, произошло ли что-нибудь, и длины очереди , чтобы убедиться, что они остались в заданных пределах.

Обновление: Чтобы объяснить, почему я почувствовал необходимость использования обеспечения ниже (меня спросили об этом в IRC), это сделано для предотвращения перекоса записи (см. Статью в Википедии о Изоляция моментальных снимков для определения). Если бы я заменил @queue на (обеспечить очередь) , два или более производителей могли бы проверить длину очереди, найти, что она меньше 4, затем поместите дополнительные элементы в очереди и, возможно, увеличивают общую длину очереди выше 4, нарушая ограничение. Точно так же два потребителя, выполняющие @queue , могут принять один и тот же элемент для обработки, а затем удалить два элемента из очереди. гарантирует, что предотвращает любой из этих сценариев.

(def go-on? (atom true))
(def queue (ref clojure.lang.PersistentQueue/EMPTY))
(def output (ref ()))
(def queue-lengths (ref ()))
(def *max-queue-length* 4)

(defn overseer
  ([] (overseer 20000))
  ([timeout]
     (Thread/sleep timeout)
     (swap! go-on? not)))

(defn queue-length-watch [_ _ _ new-queue-state]
  (dosync (alter queue-lengths conj (count new-queue-state))))

(add-watch queue :queue-length-watch queue-length-watch)

(defn producer [tag]
  (future
   (while @go-on?
     (if (dosync (let [l (count (ensure queue))]
                   (when (< l *max-queue-length*)
                     (alter queue conj tag)
                     true)))
       (Thread/sleep (rand-int 2000))))))

(defn consumer []
  (future
   (while @go-on?
     (Thread/sleep 100)       ; don't look at the queue too often
     (when-let [item (dosync (let [item (first (ensure queue))]
                               (alter queue pop)
                               item))]
       (Thread/sleep (rand-int 500))         ; do stuff
       (dosync (alter output conj item)))))) ; and let us know

(defn pro-con []
  (reset! go-on? true)
  (dorun (map #(%1 %2)
              (repeat 5 producer)
              (iterate inc 0)))
  (dorun (repeatedly 2 consumer))
  (overseer))

java.util.concurrent.LinkedBlockingQueue

Версия выше, написанная с использованием LinkedBlockingQueue . Обратите внимание, что общая схема кода в основном такая же, а некоторые детали на самом деле немного чище. Я удалил длины очереди из этой версии, поскольку LBQ позаботится об этом ограничении за нас.

(def go-on? (atom true))
(def *max-queue-length* 4)
(def queue (java.util.concurrent.LinkedBlockingQueue. *max-queue-length*))
(def output (ref ()))

(defn overseer
  ([] (overseer 20000))
  ([timeout]
     (Thread/sleep timeout)
     (swap! go-on? not)))

(defn producer [tag]
  (future
   (while @go-on?
     (.put queue tag)
     (Thread/sleep (rand-int 2000)))))

(defn consumer []
  (future
   (while @go-on?
     ;; I'm using .poll on the next line so as not to block
     ;; indefinitely if we're done; note that this has the
     ;; side effect that nulls = nils on the queue will not
     ;; be handled; there's a number of other ways to go about
     ;; this if this is a problem, see docs on LinkedBlockingQueue
     (when-let [item (.poll queue)]
       (Thread/sleep (rand-int 500)) ; do stuff
       (dosync (alter output conj item)))))) ; and let us know

(defn pro-con []
  (reset! go-on? true)
  (dorun (map #(%1 %2)
              (repeat 5 producer)
              (iterate inc 0)))
  (dorun (repeatedly 2 consumer))
  (overseer))