Clojure 等待条件而不旋转

Question

我正在实现一种机制，让线程拥有一个包含消息的队列。队列是使用来自java.util.concurrent 的LinkedBlockingQueue 构建的。我想要实现的目标如下。

Thread with mailbox:
defn work:
    * do some stuff
    * Get the head of the queue (a message):
        - if it is "hello":
             <do some stuff>
             <recur work fn>
        - if it is "bye":
             <do some stuff>
        - if it is none of the above, add the message to the back of queue
          and restart from "Get the head of the queue"
    * <reaching this point implies terminating the thread>

我尝试实现的第一个想法是使用围绕* Get the head of the queue 的循环，如果它不匹配任何子句，则使用条件检查消息并将其添加到:else 分支中的队列中.这样做的缺点是在cond 的任何子句的主体中调用recur 总是会重复循环，而使用recur（例如，在hello 案例中）意味着重复函数（即work)。所以这不是一个选择。另一个缺点是，如果此类消息需要很长时间才能到达，线程将无限期地旋转并消耗资源。

我的下一个想法（但尚未实现）是使用未来。方案如下。

* Get all the matches I have to match (i.e., "hello" and "bye")
* Start a future and pass it the list of messages:
    * While the queue does not contain any of the messages
      recur
    * when found, return the first element that matches.
* Wait for the future to deliver.
* if it is "hello":
    <do some stuff>
    <recur work fn>
  if it is "bye":
    <do some stuff>

当我这样做时，我几乎得到了我想要的：

1. 接收"hello" 或"bye" 阻止，直到我收到其中一个。
2. 我可以创建不定数量的子句来匹配消息
3. 我已经将循环行为提取到一个 future 中，该块 每次我评估我的cond 时都有很好的副作用 我确定我有匹配的消息，不必担心重试。

我真正想要但无法想象如何实现的一件事是，在这种情况下，未来不会旋转。就目前而言，它将无限期地消耗遍历队列的宝贵 CPU 资源，而永远不会收到它正在寻找的消息之一可能是完全正常的。

也许放弃LinkedBlockedQueue 并将其换成具有方法的数据结构是有意义的，例如getEither(List<E> oneOfThese)，该方法会阻塞直到这些元素之一可用。

我的另一个想法是，如果队列中没有任何元素，则在调用wait() 的队列上执行上述getEither() 操作，这是我可能在Java 中实现的一种方式。当另一个线程将消息放入队列时，我可以调用notify()，这样每个线程都会根据他想要的消息列表检查队列。

示例

下面的代码可以正常工作。但是，它有旋转问题。这基本上是我想要实现的一个非常基本的示例。

(def queue (ref '()))

(defn contains-element [elements collection]
  (some (zipmap elements (repeat true)) collection))

(defn has-element
  [col e]
  (some #(= e %) col))

(defn find-first
         [f coll]
         (first (filter f coll)))

; This function is blocking, which is what I want.
; However, it spins and thus used a LOT of cpu,
; whit is *not* what I want..
(defn get-either
  [getthese queue]
  (dosync
    (let [match   (first (filter #(has-element getthese %) @queue))
          newlist (filter #(not= match %)  @queue)]

      (if (not (nil? match))
        (do (ref-set queue newlist)
            match)
        (Thread/sleep 500)
        (recur)))))

(defn somethread
  [iwantthese]
  (let [element (get-either iwantthese queue)
        wanted  (filter #(not= % element) iwantthese)]
    (println (str "I got " element))
    (Thread/sleep 500)
    (recur wanted)))

(defn test
  []
  (.start (Thread. (fn [] (somethread '(3 4 5)))))

  (dosync (alter queue #(cons 1 %)))
  (println "Main: added 1")
  (Thread/sleep 1000)

  (dosync (alter queue #(cons 2 %)))
  (println "Main: added 2")
  (Thread/sleep 1000)

  (dosync (alter queue #(cons 3 %)))
  (println "Main: added 3")
  (Thread/sleep 1000)

  (dosync (alter queue #(cons 4 %)))
  (println "Main: added 4")
  (Thread/sleep 1000)

  (dosync (alter queue #(cons 5 %)))
  (println "Main: added 5")        
  )

有什么建议吗？

（如果有人注意到，是的，这就像演员，目的是为了学术目的在 Clojure 中实现）

Answer 1

您是否考虑过使用core.async？它以轻量级的方式提供您需要的东西。

Answer 2

您需要 2 个队列而不是 1 个：传入队列和一个“死信”队列。

1. “线程”应以阻塞方式（LinkedBlockingQueue.take()、core.async/<! 或使用代理）从传入队列中读取。
2. 如果消息不匹配任何子句：
   1. 将消息放在死队列的末尾
   2. 转到 1。
3. 如果消息匹配一个子句：
   1. 运行子句工作
   2. 对于死队列中的每条消息，匹配子句，删除匹配的子句。
   3. 转到 1。

两种实现见下文。

代理

代理与参与者非常相似，“唯一”的区别是您将数据/消息发送给参与者，但您将功能发送给代理。一个可能的实现是：

(defn create-actor [behaviour]
  (agent {:dead-queue [] 
          :behaviour behaviour}))

dead-queue 将包含不匹配任何子句的消息。这基本上是您的“队列末尾”。 behaviour 应该是 match-fn 到 fn 的某个映射/向量以运行。在我的特定实现中，我选择了一个映射，其中键是要匹配的元素，值是新项目匹配时要运行的 fn：

(def actor (create-actor {3 println
                          4 (partial println "Got a ")
                          5 #(println "Got a " %)}))

您可能需要更复杂的behaviour 数据结构。唯一重要的是知道元素是否被处理，这样你就知道元素是否必须进入死队列。

向actor发送消息：

(defn push [actor message]
  (send actor
        (fn [state new-message]
          (if-let [f (get-in state [:behaviour new-message])]
            (do
              (f new-message)
              state)
            (update-in state [:dead-queue] conj new-message)))
        message))

因此，如果behaviour 上有匹配项，则会立即处理该消息。如果不是，则将其存储在死队列中。如果您期望 behaviours 不是纯函数，您可以在处理新消息后尝试匹配/处理死队列中的所有消息。在这个示例实现中这是不可能的。

我们可以更改actor的behaviour，让死队列上的消息有机会被处理：

(defn change-behaviour [actor behaviour]
  (send actor
        (fn [state new-behaviour]
          (let [to-process (filter new-behaviour (:dead-queue state))
                new-dead-queue (vec (remove (set to-process) (:dead-queue state)))]
            (doseq [old-message to-process
                    :let [f (get new-behaviour old-message)]]
              (f old-message))
            {:behaviour new-behaviour
             :dead-queue new-dead-queue}))
        conds))

还有一个使用它的例子：

(push actor 4)
(push actor 18)
(push actor 1)
(push actor 18)
(push actor 5)
(change-behaviour actor {18 (partial println "There was an")})

和基于core.async的相同解决方案：

(defn create-actor [behaviour]
  (let [queue (async/chan)]
    (async/go-loop [dead-queue []
                    behaviour behaviour]
    (let [[type val] (async/<! queue)]
      (if (= type :data)
        (if-let [f (get behaviour val)]
          (do
            (f val)
            (recur dead-queue behaviour))
          (recur (conj dead-queue val) behaviour))
        (let [to-process (filter val dead-queue)
              new-dead-queue (vec (remove (set to-process) dead-queue))]
          (doseq [old-msg to-process
                  :let [f (get val old-msg)]]
            (f old-msg))
          (recur new-dead-queue val)))))
  queue))

(defn push [actor message]
  (async/go
    (async/>! actor [:data message])))

(defn change-behaviour [actor behaviour]
  (async/go
    (async/>! actor [:behaviour behaviour])))