本文主要向大家介绍了【云计算】openTSDB详细介绍Deferred类，通过具体的内容向大家展现，希望对大家学习云计算有所帮助。

openTSDB详解之Deferred类

0.前言

为了简单的异步处理，延迟结果的一个线程安全的实现。（原文：A thread-safe implementation of aDeferredresult for easy asynchronous processing.）

这个实现是基于Twisted的Python代码中的Deferred接口。

这个api是一个简单的，并且以优雅的方式去管理异步以及动态管道（处理链），没有显式的定义一个有限状态机【问1：什么是有限状态机？】。

我们都很繁忙，并且不一定总有时间去仔细阅读使用手册（RTFM:Read The Funcking Manual）。请将注意力放在你必须处理的，不变的东西上。除此之外，这里提供一个关于执行Deferred的概要：

一个Deferred像是一个future，具有与其相连的动态Callback chain（我译作：callback链/回调链） 当Deferred 结果变成可用的，callback 链将会被触发【因为callback是绑定在Deferred对象上的】 一个callback的结果将会被传递给下一个 当一个callback返回另一个Deferred时，在调用链中的下一个callback不能执行，直到其它的Deferred对象变得可用时【译者注：这个规则好像就是为了下述的Deferred`对象(B)`给Deferred对象(A)而制定的】 实际上有两个callback链。一个用于正常处理，一个用于错误处理/修复。 处理错误的Callback叫做errback。

Deferred是一种重要的构建块，可以以线程安全的方式编写易于使用的异步api

1.理解Deferred的概念

其思想就是：一个Deferred对象代表一个尚不可用的结果。 一个异步的操作（I/O,RPC···）已经开始，并且在将来传递它们的处理结果（要么成功，要么失败）给Deferred对象。在Deferred和Future对象之间的不同的关键点是有一个callback 链与Deferred对象关联，然而对于一个future对象，你需要在某些时候手动触发才能得到结果，这将提出一些问题，比如：你怎么知道这些结果是可用的？如果这些结果本身依赖于另外的future对象时该怎么办？

当你开始一个异步操作时，当操作完成时，你肯定想知道调用的返回结果。如果操作成功完成，你想你的callback使用它的结果去执行在你开启异步操作时做的事情。如果有一个错误，你想触发一些错误处理代码。

但是对于一个Deferred对象，不止是一次单个callback。你能够添加任意数目的callbacks，这些允许你有效且容易地构建复杂的处理管道以一个相对简单的，优雅的方式。

2.理解callback chain

让我们举一个例子。为了让其他人使用你的简单的远程存储服务，你正在为他们写一个客户端库（client library）。当你的用户调用你的库中的get()方法时，你想从你的远端服务检索一些数据，并将它们交给用户，但是你想用异步的方式处理这件事儿。

当你的客户端库的用户调用get()方法时，你收集一个请求，并且通过一个socket包将它发送到远端的服务器。在发送请求给socket之前，为了保持请求和这个Deferred对象之间的一个连接，你创建了一个Deferred对象，然后将它存储在某个地方，例如，在一个map中。然后你返回这个Deferred对象给用户， 这就是一旦RPC完成时， 他们为什么能够访问的到Deferred结果的原因。

或早或晚，RPC调用将会成功完成，亦或失败，并且你的socket将会变得可读（或者可能会在一个失败的事件中关闭）。我们假设现在所有事都像期待那样运行，因此socket也是可读的，所以你从socket包中读取响应（回复）。在这个时候，你提取远程调用的结果，然后你将其交给Deferred对象，这个Deferred对象是你为了这个请求而创建的（记住，你已经将它存储在某个地方，所以一旦你得到这个Deferred对象，你就能给它）。Deferred对象然后存储这个结果并且触发任何添加到这个Deferred对象的callback。期待的结果是：你的客户端库的用户，在调用你的get()方法之后，将会添加一个Callback到你给的Deferred对象上。这样，当Deferred结果变得可用时，你将使用参数的方式来调用它。【因为你已经将远程调用的结果发送到Deferred对象中】

到目前为止，我们已经解释了的东西并非超过带有一个callback的future 对象的范畴。但是Deferred对象所能做的远非如此。现在让我们假设，为了避免再次通过网络时得到相同的值， 其它一些人想在你的客户端库上构建一个缓冲层。想使用cache的用户将会调用在cache库中的get方法而不是直接调用你的客户端库。

假设缓冲库已经存储过一次调用的结果。 它将会创建一个Deferred对象，并立即交付给缓冲的结果， 并返回这个Deferred对象给用户。用户将会添加一个callback，这个callback将会立即调用因为Deferred结果已经可用。所以整个调用过程立即完成，并且完全来自同一个线程。没有上下文切换（没有涉及到其他的线程，没有I/O以及其它的东西），没有什么被阻塞，一切都完成的十分迅速。

现在让我们假设缓冲库有一个缓冲丢失， 并且需要使用上述的源客户端库去做一个远程get()方法调用。RPC被发送给远端的服务器，并且客户端库返回一个Deferred对象给缓冲库。这就是令人感到兴奋的事物了。在返回缓冲库给用户之前，缓冲库会添加它自己的callback给Deferred对象。这个callback将会从远端服务器取到结果，将其添加到cache中，并且返回。通常，用户添加他们自己的callback去处理结果。所以现在Deferred对象有两个callbacks与其相联系。

      1st callback       2nd callback
`Deferred`: add to cache  -->  user callback        

当RPC完成，源客户端库将会从传输中反序列化结果，并将其传递给Deferred对象。第一个callback将会被触发，这将添加结果到缓存库。然后，无论第一个callback返回什么，都会被传递给第二个callback。结果就是：缓存的callback返回得到的响应，并不会做任何更改，所以这将会被（完整地）传递给用户自定义的callback。

现在去理解第一个callback能够返回另外的任意的值就非常重要， 并且这个值能够被传递给第二个callback。第一次听到这个逻辑稍有不可思议，但是它的确是Deferred背后的关键点。

为了解释原因，可以让事情更复杂一点儿。 假设得到请求的远端服务器是一个相对简单的，且低层次的存储服务（联想到分布式缓存），所以它仅仅使用字节数组工作，它并不在意内容是什么。所以原始的客户端库仅仅从网络中反序列化字节数组并且传递字节数组到Deferred对象中。

现在你正在编写一个高层次(higher-level library)的库，使用这个存储系统（上述的存储服务）去存储一些你自定义的对象。所以当你从服务器中得到字节数组时，你需要进一步反序列化字节数组成某种对象。你的高层次库用户不在意你使用何种远端存储系统，他们在意的事情仅仅是：异步得到这些对象。你的高层次库构建于一个源端的低层次库，这个低层次库只是用来做RPC通信。

当使用高层次库的用户调用get()方法，你调用低层次库的get()方法，这个低层次库中的get()方法将会发起一个RPC调用并且返回一个Deferred对象给高层次库。 为了进一步的从字节数组到对象的反序列化，高层次库添加第一个callback。 然后高层次库的用户添加他们自己的callback，这个callback将使用这个对象一些事情。所以现在的整个流程看上去像是这个样子。

        1st callback                    2nd callback
`Deferred`: de-serialize to an object  -->  user callback

当结果来自网络时，字节数组反序列化自socket。第一个callback【第一个callback的作用就是从 字节数组—反序列化—>object】被调用并且它的参数是原始的结果——字节数组。所以1st callback进一步地反序列化字节数组成某些对象，并将其返回。 2nd callback然后被调用并且它的参数就是之前callback的结果——即反序列化而得到的对象。

现在回到缓存库，它和高层次库毫无联系。它做的所有是：某些实现某些接口的对象，带有一个get()方法，它保持一个映射——无论接收到什么参数到被缓存给这个特殊调用的对象。由于callback 链调用的方式，可能透明地使用缓存库以及高层次库。想使用缓存的用户仅仅需要使用缓存库和高层次库。现在当他们调用缓存库的get()方法时，且有一个cache丢失时，下面的这些将会依次发生：

缓存库调用高层次库中的get()方法 高层次库调用低层次库中的get()方法 低层次库创建一个Deferred对象，发起一个RPC过程并且返回它的Deferred对象 高层次库添加它的callback（用于对象反序列化）到Deferred对象中，并且将这个对象返回 客户端端添加自定义的callback（用于缓存更新）到Deferred对象中，并将其返回

用户得到Deferred对象并且添加自己的callback，使用从数据中心检索到的对象做一些事情

             1st callback       2nd callback       3rd callback
   `Deferred`: de-serialize  -->  add to cache  -->  user callback
   result: (none available)

一旦响应返回，第一个callback被调用，它反序列化得到一个对象，并返回这个对象。目前Deferred对象的结果变成了一个反序列化的对象。Deferred对象目前的状态如下所示：

      2nd callback       3rd callback
`Deferred`:  add to cache  -->  user callback
result: de-serialized object

因为在调用链中有许多callbacks，Deferred对象使用当前的结果（反序列化而成的对象）作为参数调用下一个callback。callback添加对象到自己的缓存并且不做任何修改的返回。【译者注：这里的cache层的库的callback只做添加object到自己的缓存中，而不做任何其它的操作】

              3rd callback
   `Deferred`:  user callback
   result: de-serialized object

最终，用户的callback被调用，使用一个object作为参数（原文Finally, the user's callback is invoked with the object in argument.）

`Deferred`:  (no more callbacks)
result: (whatever the user's callback returned)

如果你认为这正在变得有趣，继续读下去， 因为你尚未读到Deferred中最有趣的地方。

3.使用Deferred构建动态处理链

让我们将之前的例子变得更加复杂一点儿。 假设为这些调用服务的远端存储服务是一个运行在很多台机器上的分布式存储服务。数据分区存储在许多节点上并且从一个节点中来回移动（由于机器损坏或者其它不知名的原因）【译者注：其实笔者正在为Deferred用于HBase的环境做铺垫】。为了执行一个get调用，低层次客户端库首先需要知道哪个服务器目前存储了请求数据。假设还有另外的服务器，同时也是分布式服务的一部分，这个服务器保存了索引以及每条数据移动的轨迹。低层次的客户端库首先需要使用第一台服务器找到数据的实际存储位置（这是第一个RPC过程）【译者注：这个第一台服务器就是HBase中-root-表所在的服务器】，然后再从存储节点中检索数据（这是第二次RPC过程）。末端用户不在意检索数据涉及两个阶段，它们仅仅想调用get()方法，当数据（一个字节数组）是可用时被返回。

这大概就是Deferred最有用的特性。当用户调用get()方法时， 低层次库将会发起第一个RPC调用到索引服务器去定位用户需要的数据的位置。当发起这个RPC去查找时，一个Deferred对象被创建。低层次库的get()代码添加了第一个callback去处理查询响应，然后返回给用户。

         1st callback       2nd callback
   `Deferred`:  index lookup  -->  user callback
   result: (none available)

最终，查找的RPC完成，然后Deferred对象得到查找响应。所以在触发第一个callback之前，Deferred对象将会是这个状态。

         1st callback       2nd callback
   `Deferred`:  index lookup  -->  user callback
   result: lookup response

第一个callback运行并且现在知道该去哪儿找最开始请求的数据了。它发起了一个get()请求到正确的存储节点上。这么做将会创建另一个Deferred对象，让我们称其为B， 它是由查询索引的callback返回。并且这就是奇迹发生的地方。现在我们所出的状态如下：

   (A)        2nd callback    |   (B)
                              |
   `Deferred`:  user callback   |   `Deferred`:  (no more callbacks)
   result: `Deferred` (B)       |   result: (none available)

由于一个callback返回一个Deferred，我们不能调用用户的callback，因为用户不想知道他们的callback接收的是一个Deferred对象，他们想接收的是一个字节数组。目前的callback得到终止，并且停止处理callback 链。这个callback链需要被恢复，无论get()调用完成了什么Deferred（原文：This callback chain needs to be resumed whenever theDeferredof the get call [(B)] completes.）。为了实现这个，一个callback被追加到其它的Deferred，这个Deferred对象将会恢复callback链的执行。

   (A)        2nd callback    |   (B)        1st callback
                              |
   `Deferred`:  user callback   |   `Deferred`:  resume (A)
   result: `Deferred` (B)       |   result: (none available) 

一旦（A）添加callback到（B）上，它会立即返回，没有阻塞一个线程或者像是这类的事情需要等待。所以接收查找请求的整个过程以及发送get的RPC请求非常迅速，没有任何阻塞。

现在当get响应从网络中返回时，通常，RPC层从字节数组中反序列化，并且将其交给（B）

   (A)        2nd callback    |   (B)        1st callback
                              |
   `Deferred`:  user callback   |   `Deferred`:  resume (A)
   result: `Deferred` (B)       |   result: byte array

(B)的唯一一个callback 正准备设置（A）的结果，并且恢复（A）的callback 链

   (A)        2nd callback    |   (B)        1st callback
                              |
   `Deferred`:  user callback   |   `Deferred`:  resume (A)
   result: byte array         |   result: byte array

【译者注：这里的使用了Deferred (B) 用于设置Deferred (A)的值】

所以现在（A）恢复了它的callback链，并且调用用户的callback 使用字节数组作为参数，而这正是用户所想要的

   (A)                        |   (B)        1st callback
                              |
   `Deferred`:  (no more cb)    |   `Deferred`:  resume (A)
   result: (return value of   |   result: byte array
            the user's cb)

然后（B）移动到它callback 链的下一个，但是并不存在下一个callback链节点，所以（B）完成了。

   (A)                        |   (B)
                              |
   `Deferred`:  (no more cb)    |   `Deferred`:  (no more cb)
   result: (return value of   |   result: byte array
            the user's cb)

读取获取响应的整个过程，恢复初始的Deferred对象以及执行第二个Deferred对象发生在同一个线程里，有序的，没有任何阻塞（前提是：用户的callback没有阻塞，它也一定不能阻塞）

我们已经做的本质上与动态构建一个隐式有限状态机去处理获取请求的生命流程是相同的。这个简单的API允许你去构建任意的复杂的处理管道，这个处理管道会动态的决定在每个管道的阶段关于下一步要做什么。

处理错误：

一个Deferred对象实际上有两个而不是一个callback 链。第一个链是正常的处理流程链，第二个则是错误处理链。Twisted将错误处理callback称为errback，所以我们在这里也保持这相同的术语。当异步处理以一个错误结束时， Deferred对象必须抛出捕获的异常而不是给出结果（或者如果没有异常被捕捉，一个异常必须被创建并且传递给Deferred对象）。当目前的Deferred结果是一个异常的实例时，下一个errback将会被调用。至于正常的callbacks，无论errback返回什么，都会是目前的结果。如果当前的结果仍然是一个异常的实例，那么下一个errback将会被调用。如果当前的结果不再是一个异常，下一个callback将会被调用。

当一个callback或者errback本身抛出一个异常时，它由Deferred对象捕获，并且成为当前的结果，这就意味着链中的下一个errback将会被触发，以当前的这个异常作为参数。注意Deferred对象仅仅捕获异常，并不会抛出或者是报错。

约定及规则

仔细阅读下面这个规则，因这些是你正确实现的保证

1.一个Deferred对象能够收到仅仅一个初始值 2.每次仅仅一个线程在执行callback链 3.每个由callback动作在下一个callback被调用之前发生。换句话说，如果一个callback链操作一个变量（并且没有别人操作它），是不需要同步要求的。 4.执行callback链的线程 传递初始值给Deferred对象。这个类不创建或者管理任何的线程或者执行者。 5.一旦callback被执行，Deferred对象将会失去对它的引用。 6.每个添加了callback到一个Deferred对象的方法的时间复杂度是O(1) 7.一个Deferred对象不能接收本身作为一个初始的或者是中间值 ，因为这将引发一个无线循环。 8.你必须注意不能构建一个Deferred的循环，因为这会造成一个无限的循环（幸运的是，它会因为CallbackOverflowError而报错） 9.Callbacks以及errbacks不能接收一个Deferred作为参数。这是因为它们总是接收之间的callback的结果，当这个结果作为一个Deferred对象时，我们怀疑callback链的执行直到其它的Deferred对象的结果是可用的 10.Callbacks不能接收一个异常作为参数。这是因为它们总是导致errbacks。 11.使用Deferred对象的监视器会导致死锁，所以不要使用，换句话说，写synchronized (some_Deferred) { ... }或者是任何等价的处理会让你的保证无效。原文如下：

   Using the monitor of a Deferred can lead to a deadlock, so don't use it. In other words, writing
    synchronized (some_Deferred) { ... }
    (or anything equivalent) voids your warranty.

本文由职坐标整理并发布，希望对同学们有所帮助。了解更多详情请关注职坐标大数据云计算大数据安全频道！