Twitter标准库

最后更新于:2022-04-01 03:06:05

Twitter最重要的标准库是 [Util](http://github.com/twitter/util) 和 [Finagle](https://github.com/twitter/finagle)。Util 可以理解为Scala和Java的标准库扩展,提供了标准库中没有的功能或已有功能的更合适的实现。Finagle 是我们的RPC系统,核心分布式系统组件。 ### Future Futures已经在并发一节中简单讨论过。它是调异步处理的中心机制,渗透在我们代码库中,也是Finagle的核心。Futures允许组合并发事件,简化了高并发操作。也是JVM上异步并发的一种高效的实现。 Twitter的future是*异步*的,所以基本上任何操作(阻塞操作)——基本上任何可以suspend它的线程的执行;网络IO和磁盘IO是就是例子——必须由系统处理,它为结果提供future。Finagle为网络IO提供了这样一种系统。 Futures清晰简单:它们持有一个尚未完成运算结果的 promise 。它们是一个简单的容器——一个占位符。一次计算当然可能会失败,这种状况必须被编码:一个Future可以是三种状态之一: pending, failed, completed。 ### 闲话: *组合(composition)* 让我们重新审视我们所说的组合:将简单的组件合成一个更复杂的。函数组合的一个权威的例子:给定函数 f 和 g,组合函数 (g∘f)(x) = g(f(x)) ——结果先对 x使用f函数,然后在使用g函数——用Scala来写: ~~~ val f = (i: Int) => i.toString val g = (s: String) => s+s+s val h = g compose f // : Int => String scala> h(123) res0: java.lang.String = 123123123 ~~~ 复合函数h,是个新的函数,由之前定义的f和g函数合成。 Futures是一种集合类型——它是个包含0或1个元素的容器——你可以发现他们有标准的集合方法(eg:map, filter, foreach)。因为Future的值是延迟的,结果应用这些方法中的任何一种必然也延迟;在 ~~~ val result: Future[Int] val resultStr: Future[String] = result map { i => i.toString } ~~~ 函数 { i => i.toString } 不会被调用,直到int值可用;转换集合的resultStr在可用之前也一直是待定状态。 List可以被扁平化(flattened): ~~~ val listOfList: List[List[Int]] = .. val list: List[Int] = listOfList.flatten ~~~ 这对future也是有意义的: ~~~ val futureOfFuture: Future[Future[Int]] = .. val future: Future[Int] = futureOfFuture.flatten ~~~ 因为future是延迟的,flatten的实现——立即返回——不得不返回一个等待外部future (`**Future[**Future[Int]**]**`) 完成的future (`Future[**Future[Int]**]`).如果外部future失败,内部flattened future也将失败。 Future (类似List) 也定义了flatMap;Future[A] 定义方法flatMap的签名 ~~~ flatMap[B](f: A => Future[B]): Future[B] ~~~ 如同组合 map 和 flatten,我们可以这样实现: ~~~ def flatMap[B](f: A => Future[B]): Future[B] = { val mapped: Future[Future[B]] = this map f val flattened: Future[B] = mapped.flatten flattened } ~~~ 这是一种有威力的组合!使用flatMap我们可以定义一个 Future 作为两个Future序列的结果。第二个future 的计算基于第一个的结果。想象我们需要2次RPC调用来验证一个用户身份,我们可以用下面的方式组合操作: ~~~ def getUser(id: Int): Future[User] def authenticate(user: User): Future[Boolean] def isIdAuthed(id: Int): Future[Boolean] = getUser(id) flatMap { user => authenticate(user) } ~~~ 这种组合类型的一个额外的好处是错误处理是内置的:如果getUser(..)或authenticate(..)失败,future 从 isAuthred(..)返回时将会失败。这里我们没有额外的错误处理的代码。 #### 风格 Future回调方法(respond, onSuccess, onFailure, ensure) 返回一个新的Future,并链接到调用者。这个Future被保证只有在它调用者完成后才完成,使用模式如下: ~~~ acquireResource() future onSuccess { value => computeSomething(value) } ensure { freeResource() } ~~~ freeResource() 被保证只有在 computeSomething之后才执行,这样就模拟了try-finally 模式。 使用 onSuccess替代 foreach —— 它与 onFailure 方法对称,命名的意图更明确,并且也允许 chaining。 永远避免直接创建Promise实例: 几乎每一个任务都可以通过使用预定义的组合子完成。这些组合子确保错误和取消是可传播的, 通常鼓励的数据流风格的编程,不再需要同步和volatility声明。 用尾递归风格编写的代码不再导致堆栈空间泄漏,并使得以数据流风格高效的实现循环成为可能: ~~~ case class Node(parent: Option[Node], ...) def getNode(id: Int): Future[Node] = ... def getHierarchy(id: Int, nodes: List[Node] = Nil): Future[Node] = getNode(id) flatMap { case n@Node(Some(parent), ..) => getHierarchy(parent, n :: nodes) case n => Future.value((n :: nodes).reverse) } ~~~ Future定义很多有用的方法: 使用 Future.value() 和 Future.exception() 来创建未满意(pre-satisfied) 的future。Future.collect(), Future.join() 和 Future.select() 提供了组合子将多个future合成一个(例如:scatter-gather操作的gather部分)。 #### Cancellation Future实现了一种弱形式的取消。调用Future#cancel 不会直接终止运算,而是发送某个级别的可被任何处理查询的触发信号,最终满足这个future。Cancellation信号流向相反的方向:一个由消费者设置的cancellation信号,会传播到它的生产者。生产者使用 Promise的onCancellation来监听信号并执行相应的动作。 这意味这cancellation语意上依赖生产者,没有默认的实现。cancellation只是一个提示。 #### Local Util的[Local](https://github.com/twitter/util/blob/master/util-core/src/main/scala/com/twitter/util/Local.scala#L40)提供了一个位于特定的future派发树(dispatch tree)的引用单元(cell)。设定一个local的值,使这个值可以用于被同一个线程的Future 延迟的任何计算。有一些类似于thread locals(注:Java中的线程机制),不同的是它们的范围不是一个Java线程,而是一个 future 线程树。在 ~~~ trait User { def name: String def incrCost(points: Int) } val user = new Local[User] ... user() = currentUser rpc() ensure { user().incrCost(10) } ~~~ 在 ensure块中的 user() 将在回调被添加的时候引用 user local的值。 就thread locals来说,我们的Locals非常的方便,但要尽量避免使用:除非确信通过显式传递数据时问题不能被充分的解决,哪怕解决起来有些繁重。 Locals有效的被核心库使用在非常常见的问题上——线程通过RPC跟踪,传播监视器,为future的回调创建stack traces——任何其他解决方法都使得用户负担过度。Locals在几乎任何其他情况下都不适合。 ### Offer/Broker 并发系统由于需要协调访问数据和资源而变得复杂。[Actor](http://www.scala-lang.org/api/current/scala/actors/Actor.html)提出一种简化的策略:每一个actor是一个顺序的进程(process),保持自己的状态和资源,数据通过消息的方式与其它actor共享。 共享数据需要actor之间通信。 Offer/Broker 建立于Actor之上,以这三种重要的方式表现:1,通信通道(Brokers)是first class——即发送消息需要通过Brokers,而非直接到actor。2, Offer/Broker 是一种同步机制:通信会话是同步的。 这意味我们可以用 Broker做为协调机制:当进程a发送一条信息给进程b;a和b都要对系统状态达成一致。3, 最后,通信可以选择性地执行:一个进程可以提出几个不同的通信,其中的一个将被获取。 为了以一种通用的方式支持选择性通信(以及其他组合),我们需要将通信的描述和执行解耦。这正是Offer做的——它是一个持久数据用于描述一次通信;为了执行这个通信(offer执行),我们通过它的sync()方法同步 ~~~ trait Offer[T] { def sync(): Future[T] } ~~~ 返回 Future[T] 当通信被获取的时候生成交换值。 Broker通过offer协调值的交换——它是通信的通道: ~~~ trait Broker[T] { def send(msg: T): Offer[Unit] val recv: Offer[T] } ~~~ 所以,当创建两个offer ~~~ val b: Broker[Int] val sendOf = b.send(1) val recvOf = b.recv ~~~ sendOf和recvOf都同步 ~~~ // In process 1: sendOf.sync() // In process 2: recvOf.sync() ~~~ 两个offer都获取并且值1被交换。 通过将多个offer和Offer.choose绑定来执行可选择通信。 ~~~ def choose[T](ofs: Offer[T]*): Offer[T] ~~~ 上面的代码生成一个新的offer,当同步时获取一个特定的ofs——第一个可用的。当多个都立即可用时,随机获取一个。 Offer对象有些一次性的Offers用于与来自Broker的Offer构建。 ~~~ Offer.timeout(duration): Offer[Unit] ~~~ offer在给定时间后激活。Offer.never将用于不会有效,Offer.const(value)在给定值后立即有效。这些操作由选择性通信来组合是非常有用的。例如,在一个send操作中使用超时: ~~~ Offer.choose( Offer.timeout(10.seconds), broker.send("my value") ).sync() ~~~ 人们可能会比较 Offer/Broker 与[SynchronousQueue](http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/SynchronousQueue.html),他们有细微但非常重要的区别。Offer可以被组合,而queue不能。例如,考虑一组queues,描述为 Brokers: ~~~ val q0 = new Broker[Int] val q1 = new Broker[Int] val q2 = new Broker[Int] ~~~ 现在让我们为读取创建一个合并的queue ~~~ val anyq: Offer[Int] = Offer.choose(q0.recv, q1.recv, q2.recv) ~~~ anyq是一个将从第一个可用的queue中读取的offer。注意 anyq 仍是同步的——我们仍然拥有底层队列的语义。这类组合是不可能用queue实现的。 #### 例子:一个简单的连接池 连接池在网络应用中很常见,并且它们的实现常常需要技巧——例如,在从池中获取一个连接的时候,通常需要超时机制,因为不同的客户端有不同的延迟需求。池的简单原则:维护一个连接队列,满足那些进入的等待者。使用传统的同步原语,这通常需要两个队列(queues):一个用于等待者(当没有连接可用时),一个用于连接(当没有等待者时)。 使用 Offer/Brokers ,可以表达得非常自然: ~~~ class Pool(conns: Seq[Conn]) { private[this] val waiters = new Broker[Conn] private[this] val returnConn = new Broker[Conn] val get: Offer[Conn] = waiters.recv def put(c: Conn) { returnConn ! c } private[this] def loop(connq: Queue[Conn]) { Offer.choose( if (connq.isEmpty) Offer.never else { val (head, rest) = connq.dequeue waiters.send(head) { _ => loop(rest) } }, returnConn.recv { c => loop(connq enqueue c) } ).sync() } loop(Queue.empty ++ conns) } ~~~ loop总是提供一个归还的连接,但只有queue非空的时候才会send。 使用持久化队列(persistent queue)更进一步简化逻辑。与连接池的接口也是通过Offer实现,所以调用者如果愿意设置timeout,他们可以通过利用组合子(combinators)来做: ~~~ val conn: Future[Option[Conn]] = Offer.choose( pool.get { conn => Some(conn) }, Offer.timeout(1.second) { _ => None } ).sync() ~~~ 实现timeout不需要额外的记账(bookkeeping);这是因为Offer的语义:如果Offer.timeout被选择,不会再有offer从池中获得——连接池和它的调用者在各自waiter的broker上不必同时同意接受和发送。 #### 埃拉托色尼筛子(Sieve of Eratosthenes 译注:一种用于筛选素数的算法) 把并发程序构造为一组顺序的同步通信进程,通常很有用——有时程序被大大地简化了。Offer和Broker提供了一组工具来让它简单并一致。确实,它们的应用超越了我们可能认为是经典并发性问题——并发编程(有Offer/Broker的辅助)是一种有用的构建工具,正如子例程(subroutines),类,和模块都是——来自CSP(译注:Communicating sequential processes的缩写,即通信顺序进程)的重要思想。 这里有一个[埃拉托色尼筛子](http://ja.wikipedia.org/wiki/%E3%82%A8%E3%83%A9%E3%83%88%E3%82%B9%E3%83%86%E3%83%8D%E3%82%B9%E3%81%AE%E7%AF%A9)可以构造为一个针对一个整数流(stream of integers)的连续的应用过滤器 。首先,我们需要一个整数的源(source of integers): ~~~ def integers(from: Int): Offer[Int] = { val b = new Broker[Int] def gen(n: Int): Unit = b.send(n).sync() ensure gen(n + 1) gen(from) b.recv } ~~~ integers(n) 方法简单地提供了从n开始的所有连续的整数。然后我们需要一个过滤器: ~~~ def filter(in: Offer[Int], prime: Int): Offer[Int] = { val b = new Broker[Int] def loop() { in.sync() onSuccess { i => if (i % prime != 0) b.send(i).sync() ensure loop() else loop() } } loop() b.recv } ~~~ filter(in, p) 方法返回的offer删除了in中的所有质数(prime)的倍数。最终我们定义了我们的筛子(sieve): ~~~ def sieve = { val b = new Broker[Int] def loop(of: Offer[Int]) { for (prime <- of.sync(); _ <- b.send(prime).sync()) loop(filter(of, prime)) } loop(integers(2)) b.recv } ~~~ loop() 工作很简单:从of中读取下一个质数,然后对of应用过滤器排除这个质数。loop不断的递归,持续的质数被过滤,于是我们得到了筛选结果。我们现在打印前10000个质数: ~~~ val primes = sieve 0 until 10000 foreach { _ => println(primes.sync()()) } ~~~ 除了构造简单,组件正交,这种做法也给你一种流式筛子(streaming sieve):你不需要事先计算出你感兴趣的质数集合,从而进一步提高了模块化。
';