Clojure:替代使用互斥锁/锁和计数器

Question

场景:我有一台服务器正在监听六个活动的TCP/IP连接.当"就绪"消息进入时,将在其自己的线程上引发事件.当服务器从每个连接收到"就绪"消息时,它需要运行"启动"功能.

我面向对象的解决方案可能涉及使用互斥锁和计数器.就像是:

int _countDown= 6;
object _lock;
void ReadyMessageReceivedForTheFirstTimeFromAConnection() {
    lock(_lock) {
      --_countDown; // 
       if (_countDown==0) Start();
    }
}

如果不诉诸锁/互斥锁,怎么能在Clojure中解决这个问题呢？

Answer 1

您可以使用CountDownLatch或Phaser来实现此目的.

在我的期货图书馆,即将到来,我使用了两者.首先使用CountDownLatch将其替换为Phaser for ForkJoin兼容性(在您的情况下可能不是必需的).你可以看到这个差异的变化.希望它能让您了解两者的用法.

对于锁存器,一般的想法是:

(let [latch (CountDownLatch. 6)]
  (on-receive-message this (fn [_] (.countDown latch)))
  (.await latch)

...或类似的东西.

Answer 2

当你更喜欢纯粹的clojure版本时,你可以使用一个承诺给你的未来.

每当您收到消息时,您都会增加conn-count,手表会检查是否达到阈值并执行:转到屏障保证.

(def wait-barrier (promise))
(def conn-count (atom 0))

(add-watch conn-count :barrier-watch
           (fn [key ref old-state new-state]
             (when (== new-state 6)
               (deliver wait-barrier :go))))  

虚拟 - 例如:

(def wait-barrier (promise))
(def conn-count (atom 0))
(defn worker-dummy []
  (when (= @wait-barrier :go)
    (println "I'm a worker")))

(defn dummy-receive-msg []
  (doall (repeatedly 6,
                     (fn []
                       (println "received msg")
                       (swap! conn-count inc)))))

(let [workers (doall (repeatedly 6 (fn [] (future (worker-dummy)))))]
  (add-watch conn-count :barrier-watch
             (fn [key ref old-state new-state]
               (when (== new-state 6)
                 (deliver wait-barrier :go))))
  (dummy-receive-msg)
  (doall (map deref workers)))