Clojure的并发(八)future、promise和线程

    xiaoxiao2024-03-14  15

    Clojure 的并发(一) Ref和STM Clojure 的并发(二)Write Skew分析 Clojure 的并发(三)Atom、缓存和性能 Clojure 的并发(四)Agent深入分析和Actor Clojure 的并发(五)binding和let Clojure的并发(六)Agent可以改进的地方 Clojure的并发(七)pmap、pvalues和pcalls Clojure的并发(八)future、promise和线程  八、future、promise和线程 1 、Clojure中使用future是启动一个线程,并执行一系列的表达式,当执行完成的时候,线程会被回收: user =>  (def myfuture (future ( +   1   2 ))) # ' user/myfuture user =>  @myfuture 3 future接受一个或者多个表达式,并将这些表达式交给一个线程去处理,上面的(+ 1 2)是在另一个线程计算的,返回的future对象可以通过deref或者@宏来阻塞获取计算的结果。 future函数返回的结果可以认为是一个类似java.util.concurrent.Future的对象,因此可以取消: user =>  (future - cancelled ?  myfuture) false user =>  (future - cancel myfuture) false 也可以通过谓词future?来判断一个变量是否是future对象: user =>  (future ?  myfuture) true 2、Future的实现,future其实是一个宏,它内部是调用future-call函数来执行的: (defmacro future   [ &  body] `(future - call (fn []  ~ @body))) 可以看到,是将body包装成一个匿名函数交给future-call执行,future-call接受一个Callable对象: (defn future - call    [ ^ Callable f]   (let [fut (.submit clojure.lang.Agent / soloExecutor f)]     (reify       clojure.lang.IDeref        (deref [_] (.get fut))      java.util.concurrent.Future       (get [_] (.get fut))       (get [_ timeout unit] (.get fut timeout unit))       (isCancelled [_] (.isCancelled fut))       (isDone [_] (.isDone fut))       (cancel [_ interrupt ? ] (.cancel fut interrupt ? )))))i 将传入的Callable对象f提交给Agent的soloExecuture final   public   static  ExecutorService soloExecutor  =  Executors.newCachedThreadPool(); 执行,返回的future对象赋予fut,接下来是利用clojure 1.2引入的reify定义了一个匿名的数据类型,它有两种protocol:clojure.lang.IDeref和java.utill.concurrent.Future。其中IDeref定义了deref方法,而Future则简单地将一些方法委托给fut对象。protocol你可以理解成java中的接口,这里就是类似多态调用的作用。 这里有个地方值的学习的是,clojure定义了一个future宏,而不是直接让用户使用future-call,这符合使用宏的规则: 避免匿名函数。因为如果让用户使用future-call,用户需要将表达式包装成匿名对象传入,而提供一个宏就方便许多。 3、启动线程的其他方法,在clojure中完全可以采用java的方式去启动一个线程: user =>  (.start (Thread. #(println  " hello " ))) nil hello 4、promise用于线程之间的协调通信,当一个promise的值还没有设置的时候,你调用deref或者@想去解引用的时候将被阻塞: user =>  (def mypromise (promise)) # ' user/mypromise user =>  @mypromise 在REPL执行上述代码将导致REPL被挂起,这是因为mypromise还没有值,你直接调用了@mypromise去解引用导致主线程阻塞。 如果在调用@宏之前先给promise设置一个值的话就不会阻塞: user =>  (def mypromise (promise)) # ' user/mypromise user =>  (deliver mypromise  5 ) # < AFn$IDeref$db53459f@c0f1ec:  5 > user =>  @mypromise                5 通过调用deliver函数给mypromise传递了一个值,这使得后续的@mypromise直接返回传递的值5。显然promise可以用于不同线程之间的通信和协调。 5、promise的实现:promise的实现非常简单,是基于CountDownLatch做的实现,内部除了关联一个CountDownLatch还关联一个atom用于存储值: (defn promise   []   (let [d (java.util.concurrent.CountDownLatch.  1 )         v (atom nil)]     (reify       clojure.lang.IDeref       (deref [_] (.await d) @v)      clojure.lang.IFn       (invoke [ this  x]         (locking d           ( if  (pos ?  (.getCount d))             ( do  (reset !  v x)                 (.countDown d)                  this )             ( throw  (IllegalStateException.  " Multiple deliver calls to a promise " )))))))) d是一个CountDownLatch,v是一个atom,一开始值是nil。返回的promise对象也是通过reify定义的匿名数据类型,他也是有两个protocol,一个是用于deref的IDeref,简单地调用d.await()阻塞等待;另一个是匿名函数,接受两个参数,第一个是promise对象自身,第二个参数是传入的值x,当d的count还大于0的请看下,设置v的值为x,否则抛出异常的多次deliver了。查看下deliver函数,其实就是调用promise对象的匿名函数protocol: (defn deliver   {:added  " 1.1 " }   [promise val] (promise val)) 文章转自庄周梦蝶  ,原文发布时间 2010-08-08
    最新回复(0)