【发布时间】：2013-05-27 22:25:54
【问题描述】：
对于这篇笔记的篇幅，我提前道歉。我花了相当多的时间使它更短，而且这是我能得到的最小的。
我有一个谜，非常感谢您的帮助。这个谜团来自一个 rxjava observer 的行为，我在 Clojure 中写了几个简单的 observables 抄自在线示例。
一个 observable 同步向其观察者的 onNext 处理程序发送消息，而我所谓的有原则的观察者的行为符合预期。
另一个 observable 在另一个线程上通过 Clojure future 异步执行相同的操作。完全相同的观察者不会捕获发布到其onNext 的所有事件；它似乎只是在尾部丢失了随机数量的消息。
在promised onCompleted 的等待到期和发送到agent 收集器的所有事件的等待到期之间存在故意竞争。如果promise 获胜，我希望看到false 对应onCompleted，并且agent 中的队列可能很短。如果agent 获胜，我希望看到true 对应onCompleted 以及来自agent 队列的所有消息。我不期望的一个结果是true for onCompleted 和来自agent 的短队列。但是，墨菲不睡觉，而这正是我所看到的。我不知道垃圾收集是否有问题，或者 Clojure 的 STM 内部排队，或者我的愚蠢，或者其他什么。
我在这里按照自包含形式的顺序呈现源代码，以便它可以直接通过lein repl 运行。需要避开三个仪式：首先，leiningen 项目文件project.clj，它声明了对Netflix rxjava 的0.9.0 版本的依赖：
(defproject expt2 "0.1.0-SNAPSHOT"
:description "FIXME: write description"
:url "http://example.com/FIXME"
:license {:name "Eclipse Public License"
:url "http://www.eclipse.org/legal/epl-v10.html"}
:dependencies [[org.clojure/clojure "1.5.1"]
[com.netflix.rxjava/rxjava-clojure "0.9.0"]]
:main expt2.core)
现在，命名空间和 Clojure 要求以及 Java 导入：
(ns expt2.core
(:require clojure.pprint)
(:refer-clojure :exclude [distinct])
(:import [rx Observable subscriptions.Subscriptions]))
最后是输出到控制台的宏：
(defmacro pdump [x]
`(let [x# ~x]
(do (println "----------------")
(clojure.pprint/pprint '~x)
(println "~~>")
(clojure.pprint/pprint x#)
(println "----------------")
x#)))
最后，我的观察者。我使用 agent 来收集任何 observable 的 onNext 发送的消息。我使用atom 来收集潜在的onError。我将promise 用于onCompleted，以便观察者外部的消费者可以等待它。
(defn- subscribe-collectors [obl]
(let [;; Keep a sequence of all values sent:
onNextCollector (agent [])
;; Only need one value if the observable errors out:
onErrorCollector (atom nil)
;; Use a promise for 'completed' so we can wait for it on
;; another thread:
onCompletedCollector (promise)]
(letfn [;; When observable sends a value, relay it to our agent"
(collect-next [item] (send onNextCollector (fn [state] (conj state item))))
;; If observable errors out, just set our exception;
(collect-error [excp] (reset! onErrorCollector excp))
;; When observable completes, deliver on the promise:
(collect-completed [ ] (deliver onCompletedCollector true))
;; In all cases, report out the back end with this:
(report-collectors [ ]
(pdump
;; Wait for everything that has been sent to the agent
;; to drain (presumably internal message queues):
{:onNext (do (await-for 1000 onNextCollector)
;; Then produce the results:
@onNextCollector)
;; If we ever saw an error, here it is:
:onError @onErrorCollector
;; Wait at most 1 second for the promise to complete;
;; if it does not complete, then produce 'false'.
;; I expect if this times out before the agent
;; times out to see an 'onCompleted' of 'false'.
:onCompleted (deref onCompletedCollector 1000 false)
}))]
;; Recognize that the observable 'obl' may run on another thread:
(-> obl
(.subscribe collect-next collect-error collect-completed))
;; Therefore, produce results that wait, with timeouts, on both
;; the completion event and on the draining of the (presumed)
;; message queue to the agent.
(report-collectors))))
现在，这是一个同步的 observable。它向观察者的onNext 发送 25 条消息，然后调用他们的onCompleteds。
(defn- customObservableBlocking []
(Observable/create
(fn [observer] ; This is the 'subscribe' method.
;; Send 25 strings to the observer's onNext:
(doseq [x (range 25)]
(-> observer (.onNext (str "SynchedValue_" x))))
; After sending all values, complete the sequence:
(-> observer .onCompleted)
; return a NoOpSubsription since this blocks and thus
; can't be unsubscribed (disposed):
(Subscriptions/empty))))
我们为我们的观察者订阅这个 observable：
;;; The value of the following is the list of all 25 events:
(-> (customObservableBlocking)
(subscribe-collectors))
它按预期工作，我们在控制台上看到以下结果
{:onNext (do (await-for 1000 onNextCollector) @onNextCollector),
:onError @onErrorCollector,
:onCompleted (deref onCompletedCollector 1000 false)}
~~>
{:onNext
["SynchedValue_0"
"SynchedValue_1"
"SynchedValue_2"
"SynchedValue_3"
"SynchedValue_4"
"SynchedValue_5"
"SynchedValue_6"
"SynchedValue_7"
"SynchedValue_8"
"SynchedValue_9"
"SynchedValue_10"
"SynchedValue_11"
"SynchedValue_12"
"SynchedValue_13"
"SynchedValue_14"
"SynchedValue_15"
"SynchedValue_16"
"SynchedValue_17"
"SynchedValue_18"
"SynchedValue_19"
"SynchedValue_20"
"SynchedValue_21"
"SynchedValue_22"
"SynchedValue_23"
"SynchedValue_24"],
:onError nil,
:onCompleted true}
----------------
这是一个异步 observable，它做的事情完全相同，只是在 future 的线程上：
(defn- customObservableNonBlocking []
(Observable/create
(fn [observer] ; This is the 'subscribe' method
(let [f (future
;; On another thread, send 25 strings:
(doseq [x (range 25)]
(-> observer (.onNext (str "AsynchValue_" x))))
; After sending all values, complete the sequence:
(-> observer .onCompleted))]
; Return a disposable (unsubscribe) that cancels the future:
(Subscriptions/create #(future-cancel f))))))
;;; For unknown reasons, the following does not produce all 25 events:
(-> (customObservableNonBlocking)
(subscribe-collectors))
但是，令人惊讶的是，这是我们在控制台上看到的：true for onCompleted，这意味着 promise DID NOT TIME-OUT；但只有一些异步消息。我们看到的实际消息数量因运行而异，这意味着存在一些并发现象。线索表示赞赏。
----------------
{:onNext (do (await-for 1000 onNextCollector) @onNextCollector),
:onError @onErrorCollector,
:onCompleted (deref onCompletedCollector 1000 false)}
~~>
{:onNext
["AsynchValue_0"
"AsynchValue_1"
"AsynchValue_2"
"AsynchValue_3"
"AsynchValue_4"
"AsynchValue_5"
"AsynchValue_6"],
:onError nil,
:onCompleted true}
----------------
【问题讨论】：
标签： clojure reactive-programming netflix rx-java