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

Question

>场景：我有一台服务器侦听六个活动的TCP/IP连接。当出现"就绪"消息时，事件将在其自己的线程上引发。当服务器从每个连接收到"就绪"消息时，它需要运行"启动"功能。

我的面向对象解决方案可能涉及使用互斥锁和计数器。像这样：

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

如何在不诉诸锁/互斥锁的情况下在 Clojure 中解决这个问题？

Answer 1

当你喜欢一个纯粹的clojure版本时，你可以用一个承诺来试一试你的未来。

每次收到消息时，您都会增加连接计数手表检查是否达到侵入，并交付：转到障碍承诺。

(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)))

Answer 2

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

在我的未来图书馆中，迫在眉睫，我同时使用了两者。首先使用CountDownLatch，然后将其替换为Phaser以实现ForkJoin兼容性（在您的情况下可能不是必需的）。您可以在此差异中看到更改。希望它能让您了解两者的用法。

使用闩锁，一般思路是：

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

。或类似的东西。

Answer 3

由于到目前为止还没有提到它：你可以用core.async轻松地做到这一点。看看这个MCVE：

(let [conn-count 6
      ready-chan (chan)]
  ;; Spawn a thread for each connection.
  (doseq [conn-id (range conn-count)]
    (thread
      (Thread/sleep (rand-int 2000))
      (>!! ready-chan conn-id)))
  ;; Block until all connections are established.
  (doseq [total (range 1 (inc conn-count))]
    (println (<!! ready-chan) "connected," total "overall"))
  ;; Invoke start afterwards.
  (println "start"))
;; 5 connected, 1 overall
;; 3 connected, 2 overall
;; 4 connected, 3 overall
;; 0 connected, 4 overall
;; 1 connected, 5 overall
;; 2 connected, 6 overall
;; start
;;=> nil

您还可以使用通道来实现倒计时锁存器（借用自Christophe Grand）：

(defn count-down-latch-chan [n]
  (chan 1 (comp (drop (dec n)) (take 1))))

有关core.async的简短介绍，请查看此要点。如需更长的内容，请阅读"勇敢与真实的Clojure"中的相应章节。