Clojure

代理和异步操作

与 Refs 类似,代理提供对可变状态的共享访问。其中 Refs 支持对 多个 位置进行 协调 的、同步 的更改,而代理则提供对 单个 位置进行 独立 的、异步 的更改。代理在其生命周期内绑定到一个存储位置,并且只允许作为操作结果对该位置(更改为新状态)进行修改。操作是函数(可选地,带有一些附加参数),这些函数会异步应用于代理的状态,其返回值将成为代理的新状态。由于操作是函数,因此它们也可以是多方法,因此操作具有潜在的多态性。此外,由于函数集是开放的,因此代理支持的操作集也是开放的,这与某些其他语言提供的模式匹配消息处理循环形成了鲜明对比。

Clojure 的代理是 反应式 的,而不是自主的 - 没有命令式消息循环,也没有阻塞接收。代理的状态本身应该是不变的(最好是 Clojure 持久集合的实例之一),并且任何线程都可以立即读取代理的状态(使用 deref 函数或 读取器 宏 @),无需任何消息,即观察不需要协作或协调。

代理操作调度采用 (send agent fn args*) 的形式。 send(和 send-off)总是立即返回。在稍后的某个时间点,在另一个线程中,将发生以下情况

  1. 给定的 fn 将应用于代理的 状态 和 args(如果提供了任何 args)。

  2. fn 的返回值将传递给验证器函数(如果已在代理上设置)。有关详细信息,请参阅 set-validator!

  3. 如果验证器成功或未提供验证器,则给定 fn 的返回值将成为代理的新状态。

  4. 如果任何观察者已添加到代理,则将调用它们。有关详细信息,请参阅 add-watch

  5. 如果在函数执行期间进行了任何其他调度(直接或间接),它们将保留到代理的状态更改 之后

如果操作函数引发任何异常,则不会发生嵌套调度,并且异常将在代理本身中缓存。当代理缓存了错误时,任何后续交互都将立即引发异常,直到代理的错误被清除。可以使用 agent-error 检查代理错误,并使用 restart-agent 重新启动代理。

所有代理的操作都在线程池中的线程之间交错。在任何时间点,每个代理最多只有一个操作正在执行。从另一个单个代理或线程调度到代理的操作将按发送顺序发生,可能与从其他来源调度到同一代理的操作交错。 send 应用于 CPU 受限的操作,而 send-off 适用于可能阻塞 IO 的操作。

代理与 STM 集成 - 在事务中进行的任何调度都将保留到事务提交,如果事务重试或中止,则将丢弃。

与 Clojure 的所有并发支持一样,不涉及任何用户代码锁定。

请注意,使用代理会启动一个非守护进程后台线程池,这将阻止 JVM 关闭。使用 shutdown-agents 终止这些线程并允许关闭。

示例

此示例是发送消息绕环测试的实现。创建了一个包含 m 个代理的链,然后将 n 个操作的序列调度到链的头部并通过它中继。

(defn relay [x i]
  (when (:next x)
    (send (:next x) relay i))
  (when (and (zero? i) (:report-queue x))
    (.put (:report-queue x) i))
  x)

(defn run [m n]
  (let [q (new java.util.concurrent.SynchronousQueue)
        hd (reduce (fn [next _] (agent {:next next}))
                   (agent {:report-queue q}) (range (dec m)))]
    (doseq [i (reverse (range n))]
      (send hd relay i))
    (.take q)))

; 1 million message sends:
(time (run 1000 1000))
->"Elapsed time: 2959.254 msecs"

创建代理: agent

检查代理: deref (另请参阅 @ 读取器 宏) agent-error error-handler error-mode

更改代理状态: send send-off restart-agent

阻塞等待代理: await await-for

Ref 验证器: set-validator! get-validator

观察者: add-watch remove-watch

代理线程管理: shutdown-agents