Clojure中的线程同步

Question

Clojure中的线程同步

multithreadingclojureclojure-java-interop

3

我有一个练习：

Print in order all positive integers from 1 to 100.

Using blocks, semaphores or other similar mechanisms (but avoiding sleep), coordinate two threads such that the combined output from both threads appears in numerical order.

 Sample Output
    In thread one: The number is ‘1’
    In thread two: The number is ‘2’
    In thread one: The number is ‘3’
    In thread two: The number is ‘4’

这个练习是关于Ruby的，但我想向我的班级展示Clojure可能是完成任务的好选择。

虽然我没有使用任何语言中的线程的经验，但我在考虑使用类似以下的东西：

 (def thread_1 (future (swap! my-atom inc) ))
 (def thread_2 (future (swap! my-atom inc) ))

但是@thread_1总是返回相同的值。有没有一种方法在Clojure中协调两个线程？

我在Java中使用ReentrantLock和Condition找到了这个示例，现在我正在尝试将其翻译成Clojure。

- aarkerio

顺序很重要吗？我的意思是线程应该像你的例子一样一个接一个地工作。 - fl00r

1

@fl00r，是的，顺序必须是连续的。 - aarkerio

3个回答

3

首先，你看到奇怪的结果是因为你解引用了future，而不是包含数字的原子。当解引用时，future将返回swap！的结果。

其次，你可以使用locking（基本上是Java的synchronized）来一次只允许一个线程增加和打印：

(def n-atom (atom 0))

(defn action []
  ; Arbitrarily chose the atom to lock on.
  ; It would probably be better to create a private object that can't be otherwise used.
  (locking n-atom
    (println
      (swap! n-atom inc))))

(defn go []
  ; Have each thread do action twice for a total of four times
  (future (doall (repeatedly 2 action)))
  (future (doall (repeatedly 2 action))))

(go)
1
2
3
4

需要注意的是，这里真的不应该使用future。当你想要异步计算结果时，可以使用future。但它会吞噬错误，直到被解引用，所以如果你从未使用@，你将永远不会看到在future内部出现的异常。最好使用线程池，或者借助宏来方便地启动两个线程：

(defmacro thread
  "Starts the body in a new thread."
  [& body]
  `(doto (Thread. ^Runnable (fn [] ~@body))
         (.start)))

(defn go []
  (thread (doall (repeatedly 2 action)))
  (thread (doall (repeatedly 2 action))))

- Carcigenicate

3

这是一种使用代理（agent）来协调操作同一状态的多个线程的方法。对于IT技术较为生疏的人，这里是一个代理的链接。

(def my-agent (agent 1 :validator #(<= % 100)))
(future
  (while true
    (send my-agent
          (fn [i]
            (println "In thread" (.getName (Thread/currentThread))
                     "the number is:" i)
            (inc i)))))
In thread clojure-agent-send-pool-4 the number is: 1
In thread clojure-agent-send-pool-5 the number is: 2
In thread clojure-agent-send-pool-5 the number is: 3
In thread clojure-agent-send-pool-4 the number is: 4
In thread clojure-agent-send-pool-4 the number is: 5
In thread clojure-agent-send-pool-4 the number is: 6

不论使用还是不使用future，这里都会看到相同的基本行为，因为在内部循环中send立即返回，并且sent函数可能在池中的不同线程上执行。代理负责协调对共享状态的访问。 更新：以下是另一种实现相同功能的方式，它不涉及:validator函数或异常终止。

(def my-agent (agent (range 1 101)))
(while (seq @my-agent)
  (send
    my-agent
    (fn [[n & ns]]
      (when n
        (println "In thread" (.getName (Thread/currentThread))
                 "the number is:" n)
        ns))))

- Taylor Wood

2

这里的信息并没有错误，但是使用验证器来处理普通终止吗？让未来在普通终止时崩溃吗？好吧，这个例子可以改进一下。 - glts

1

@glts 说得好！请查看更新后的答案，如果您有其他改进意见，请告诉我。 - Taylor Wood

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- fl00r · Accepted Answer

如果线程顺序很重要，而且您对非经典的线程通信感到好奇，那么您可以选择使用clojure.core.async并使用“通道”来完成。

(require '[clojure.core.async :as a])

(let [chan-one (a/chan 1)
      chan-two (a/chan 1)]
  (a/>!! chan-one 1)
  (doseq [[thread in out] [["one" chan-one chan-two]
                           ["two" chan-two chan-one]]]
    (a/go-loop []
      (when-let [n (a/<! in)]
        (if (> n 10)
          (do (a/close! in)
              (a/close! out))
          (do (prn (format "In thread %s: The number is `%s`" thread n))
              (a/>! out (inc n))
              (recur)))))))

输出结果为：

"In thread one: The number is `1`"
"In thread two: The number is `2`"
"In thread one: The number is `3`"
"In thread two: The number is `4`"
"In thread one: The number is `5`"
"In thread two: The number is `6`"
"In thread one: The number is `7`"
"In thread two: The number is `8`"
"In thread one: The number is `9`"
"In thread two: The number is `10`"

这里需要注意的一点是，go-routine在一个线程池中执行，因此它们不是专用线程，如果您想使用真正的线程，应该这样做：

(require '[clojure.core.async :as a])

(let [chan-one (a/chan 1)
      chan-two (a/chan 1)]
  (a/>!! chan-one 1)
  (doseq [[thread in out] [["one" chan-one chan-two]
                           ["two" chan-two chan-one]]]
    (a/thread
      (loop []
        (when-let [n (a/<!! in)]
          (if (> n 10)
            (do (a/close! in)
                (a/close! out))
            (do (prn (format "In thread %s: The number is `%s`" thread n))
                (a/>!! out (inc n))
                (recur))))))))