亚洲国产日韩欧美一区二区三区,精品亚洲国产成人av在线,国产99视频精品免视看7,99国产精品久久久久久久成人热,欧美日韩亚洲国产综合乱

characters

Finagle-Friendly REPL

我們將要討論的不是標準Scala的代碼。如果你喜歡使用REPL學習，你可能想知道如何獲得一個加入Finagle 及其依賴的 Scala REPL。

你可以在這里獲取Finagle源代碼。

如果你在 finagle目錄下有Finagle的源代碼，你可以通過下面的命令得到一個控制臺

$ cd finagle
$ ./sbt "project finagle-http" console
 ...build output...
scala>

Futures

Finagle使用 com.twitter.util.Future1 編碼延遲操作。Future是尚未生成的值的一個句柄。Finagle使用Future作為其異步API的返回值。同步API會在返回前等待結果；但是異步API則不會等待。例如，個對互聯網上一些服務的HTTP請求可能半秒都不會返回。你不希望你的程序阻塞等待半秒。 “慢”的API可以立即返回一個Future，然后在需要解析其值時“填充”。

val myFuture = MySlowService(request) // returns right away
   ...do other things...
val serviceResult = myFuture.get() // blocks until service "fills in" myFuture

在實踐中，你不會發(fā)送一個請求，然后在幾行代碼后調用myFuture.get。Future提供了注冊回調的方法，在值變得可用時會調用注冊的回調函數。

如果你用過其他異步API，當看到“回調”你也許會畏縮。你可能會聯想到他們難以辨認的代碼流，被調用的函數藏在離調用處遠遠的地方。但是，Future可以利用Scala中“函數是一等公民”的特性編寫出更可讀的代碼流。你可以在調用它的地方簡單地定義一個處理函數。

例如，寫代碼調度請求，然后“處理”回應，你可以保持代碼在一起：

val future = dispatch(req) // returns immediately, but future is "empty"
future onSuccess { reply => // when the future gets "filled", use its value
  println(reply)
}

你可以在REPL中用體驗一下Future。雖然不是學習如何在實際代碼中使用他們的好方法，但可以幫助理解API。當你使用REPL，Promise是一個方便的類。它是Future抽象類的一個具體子類。你可以用它來創(chuàng)建一個還沒有值的Future。

scala> import com.twitter.util.{Future,Promise}
import com.twitter.util.{Future, Promise}

scala> val f6 = Future.value(6) // create already-resolved future
f6: com.twitter.util.Future[Int] = com.twitter.util.ConstFuture@c63a8af

scala> f6.get()
res0: Int = 6

scala> val fex = Future.exception(new Exception) // create resolved sad future
fex: com.twitter.util.Future[Nothing] = com.twitter.util.ConstFuture@38ddab20

scala> fex.get()
java.lang.Exception
  ... stack trace ...

scala> val pr7 = new Promise[Int] // create unresolved future
pr7: com.twitter.util.Promise[Int] = Promise@1994943491(...)

scala> pr7.get()
  ...console hangs, waiting for future to resolve...
Ctrl-C
Execution interrupted by signal.

scala> pr7.setValue(7)

scala> pr7.get()
res1: Int = 7

scala>

在實際代碼中使用Future時，你通常不會調用get ，而是使用回調函數。 get僅僅是方便在REPL修修補補。

 

順序組合（Sequential composition）

Future有類似集合API中的組合子（如 map, flatMap） 。回顧一下集合組合子，它讓你可以表達如 “我有一個整數List和一個square函數：map那個列表獲得整數平方的列表”這樣的操作。這種表達方式很靈巧；你可以把組合子函數和另一個函數放在一起有效地組成一個新函數。面向Future的組合子可以讓你這樣表達：“我有一個期望整數的Future和一個square函數：map那個Future獲得一個期望整數平方的Future”。

如果你在定義一個異步API，傳入一個請求值，你的API應該返回一個包裝在Future中的響應。因此，這些把輸入和函數加入Future的組合子是相當有用的：它們幫助你根據其它異步API定義你自己的異步API。

最重要的Future的組合子是flatMap2：

def Future[A].flatMap[B](f: A => Future[B]): Future[B]

flatMap 序列化兩個Future。即，它接受一個Future和一個異步函數，并返回另一個Future。方法簽名中是這樣寫的：給定一個Future成功的值，函數f提供下一個Future。如果/當輸入的Future 成功完成，flatMap自動調用f。只有當這兩個Future都已完成，此操作所代表的Future才算完成。如果任何一個Future失敗，則操作確定的 Future也將失敗。這種隱交織的錯誤讓我們只需要在必要時來處理錯誤，所以語法意義很大。flatMap是這些語義組合子的標準名稱。

如果你有一個Future并且想在異步API使用其值，使用flatMap。例如，假設你有一個Future[User]，需要一個Future[Boolean]表示用戶是否已被禁止。有一個isBanned 的異步API來判斷一個用戶是否已被禁止。此時可以使用flatMap ：

scala> import com.twitter.util.{Future,Promise}
import com.twitter.util.{Future, Promise}

scala> class User(n: String) { val name = n }
defined class User

scala> def isBanned(u: User) = { Future.value(false) }
isBanned: (u: User)com.twitter.util.Future[Boolean]

scala> val pru = new Promise[User]
pru: com.twitter.util.Promise[User] = Promise@897588993(...)

scala> val futBan = pru flatMap isBanned // apply isBanned to future
futBan: com.twitter.util.Future[Boolean] = Promise@1733189548(...)

scala> futBan.get()
  ...REPL hangs, futBan not resolved yet...
Ctrl-C
Execution interrupted by signal.

scala> pru.setValue(new User("prudence"))

scala> futBan.get()
res45: Boolean = false

scala>

同樣，如果要在Future中應用一個同步函數，可以使用map。例如，假設你有一個Future[RawCredentials]需要一個Future[Credentials]。你有一個的同步的normalize函數將RawCredentials轉換成Credentials?？梢允褂?code>map：

scala> class RawCredentials(u: String, pw: String) {
     |   val username = u
     |   val password = pw
     | }
defined class RawCredentials

scala> class Credentials(u: String, pw: String) {
     |   val username = u
     |   val password = pw
     | }
defined class Credentials

scala> def normalize(raw: RawCredentials) = {
     |   new Credentials(raw.username.toLowerCase(), raw.password)
     | }
normalize: (raw: RawCredentials)Credentials

scala> val praw = new Promise[RawCredentials]
praw: com.twitter.util.Promise[RawCredentials] = Promise@1341283926(...)

scala> val fcred = praw map normalize // apply normalize to future
fcred: com.twitter.util.Future[Credentials] = Promise@1309582018(...)

scala> fcred.get()
   ...REPL hangs, fcred doesn't have a value yet...
Ctrl-C
Execution interrupted by signal.

scala> praw.setValue(new RawCredentials("Florence", "nightingale"))

scala> fcred.get().username
res48: String = florence

scala>

Scala有快捷語法來調用flatMap：for表達式。假設你想通過異步API驗證登錄請求，然后通過另一個異步API檢查用戶是否被禁止。在for表達式的幫助下，我們可以這樣寫：

scala> def authenticate(req: LoginRequest) = {
     |   // TODO: we should check the password
     |   Future.value(new User(req.username))
     | }
authenticate: (req: LoginRequest)com.twitter.util.Future[User]

scala> val f = for {
     |  u <- authenticate(request)
     |  b <- isBanned(u)
     | } yield (u, b)
f: com.twitter.util.Future[(User, Boolean)] = Promise@35785606(...)

scala>

它產生一個f: Future[(User, Boolean)]，包含用戶對象和一個表示該用戶是否已被禁止的布爾值。注意這里是怎樣實現順序組合的：isBanned使用了authenticate的輸出作為其輸入。

 

并發(fā)組合

你可能想一次獲取來自多個服務的數據。例如，如果你正在編寫一個Web服務來顯示內容和廣告，它可能會從兩個服務中分別獲取內容和廣告。但是，你怎么告訴代碼來等待兩份答復呢？如果必須自己實現可能會非常棘手，幸運的是你可以使用并發(fā)組合子。

Future 提供了一些并發(fā)組合子。一般來說，他們都是將Future的一個序列轉換成包含一個序列的Future，只是方式略微不同。這是很好的，因為它（本質上）可以讓你把幾個Future封裝成一個單一的Future。

object Future {
  …
  def collect[A](fs: Seq[Future[A]]): Future[Seq[A]]
  def join(fs: Seq[Future[_]]): Future[Unit]
  def select(fs: Seq[Future[A]]) : Future[(Try[A], Seq[Future[A]])]
}

collect參數是具有相同類型Future的一個集合，返回一個Future，其類型是包含那個類型值的一個序列。當所有的Future都成功完成或者當中任何一個失敗，都會使這個Future完成。返回序列的順序和傳入序列的順序相對應。

scala> val f2 = Future.value(2)
f2: com.twitter.util.Future[Int] = com.twitter.util.ConstFuture@13ecdec0

scala> val f3 = Future.value(3)
f3: com.twitter.util.Future[Int] = com.twitter.util.ConstFuture@263bb672

scala> val f23 = Future.collect(Seq(f2, f3))
f23: com.twitter.util.Future[Seq[Int]] = Promise@635209178(...)

scala> val f5 = f23 map (_.sum)
f5: com.twitter.util.Future[Int] = Promise@1954478838(...)

scala> f5.get()
res9: Int = 5

join參數是混合類型的Future序列，返回一個Future[Unit]，當所有的相關Future完成時（無論他們是否失敗）該Future完成。其作用是標識一組異構操作完成。對那個內容和廣告的例子來說，這可能是一個很好的解決方案。

scala> val ready = Future.join(Seq(f2, f3))
ready: com.twitter.util.Future[Unit] = Promise@699347471(...)

scala> ready.get() // doesn't ret value, but I know my futures are done

scala>

當傳入的Future序列的第一個Future完成的時候，select會返回一個Future。它會將那個完成的Future和其它未完成的Future一起放在Seq中返回。 （它不會做任何事情來取消剩余的Future。你可以等待更多的回應，或者忽略他們）

scala> val pr7 = new Promise[Int] // unresolved future
pr7: com.twitter.util.Promise[Int] = Promise@1608532943(...)

scala> val sel = Future.select(Seq(f2, pr7)) // select from 2 futs, one resolved
sel: com.twitter.util.Future[...] = Promise@1003382737(...)

scala> val(complete, stragglers) = sel.get()
complete: com.twitter.util.Try[Int] = Return(2)
stragglers: Seq[...] = List(...)

scala> complete.get()
res110: Int = 2

scala> stragglers(0).get() // our list of not-yet-finished futures has one item
  ...get() hangs the REPL because this straggling future is not finished...
Ctrl-C
Execution interrupted by signal.

scala> pr7.setValue(7)

scala> stragglers(0).get()
res113: Int = 7

scala>

 

組合例子：緩存速率限制

這些組合子表達了典型的網絡服務操作。這段假設的代碼在對速率進行限制（為了保持本地速率限制緩存）的同時，將用戶的請求調度到后臺服務：

// Find out if user is rate-limited. This can be slow; we have to ask
// the remote server that keeps track of who is rate-limited.
def isRateLimited(u: User): Future[Boolean] = {
  ...
}

// Notice how you can swap this implementation out now with something that might
// implement a different, more restrictive policy.

// Check the cache to find out if user is rate-limited. This cache
// implementation is just a Map, and can return a value right way. But we
// return a Future anyhow in case we need to use a slower implementation later.
def isLimitedByCache(u: User): Future[Boolean] =  Future.value(limitCache(u))

// Update the cache
def setIsLimitedInCache(user: User, v: Boolean) { limitCache(user) = v }

// Get a timeline of tweets... unless the user is rate-limited (then throw
// an exception instead)
def getTimeline(cred: Credentials): Future[Timeline] =
  isLimitedByCache(cred.user) flatMap {
    case true => Future.exception(new Exception("rate limited"))
    case false =>

      // First we get auth'd user then we get timeline.
      // Sequential composition of asynchronous APIs: use flatMap
      val timeline = auth(cred) flatMap(getTimeline)
      val limited = isRateLimited(cred.user) onSuccess(
                                       setIsLimitedInCache(cred.user, _))

      // 'join' concurrently combines differently-typed futures
      // 'flatMap' sequentially combines, specifies what to do next
      timeline join limited flatMap {
        case (_, true) => Future.exception(new Exception("rate limited"))
        case (timeline, _) => Future.value(timeline)
      }
  }
}

這個例子結合了順序和并發(fā)組合。請注意，除了給轉化速率限制回應一個異常以外，沒有明確的錯誤處理。如果任何Future在這里失敗，它會自動傳播到返回的Future中。

 

組合例子：網絡爬蟲

你已經看到了怎樣使用Future組合子的例子，不過也許意猶未盡。假設你有一個簡單的互聯網模型。該互聯網中只有HTML網頁和圖片，其中頁面可以鏈接到圖像和其他網頁。你可以獲取一個頁面或圖像，但API是異步的。這個假設的API成這些“可獲取”的數據為資源：

import com.twitter.util.{Try,Future,Promise}

// a fetchable thing
trait Resource {
  def imageLinks(): Seq[String]
  def links(): Seq[String]
}

// HTML pages can link to Imgs and to other HTML pages.
class HTMLPage(val i: Seq[String], val l: Seq[String]) extends Resource {
  def imageLinks() = i
  def links = l
}

// IMGs don't actually link to anything else
class Img() extends Resource {
  def imageLinks() = Seq()
  def links() = Seq()
}

// profile.html links to gallery.html and has an image link to portrait.jpg
val profile = new HTMLPage(Seq("portrait.jpg"), Seq("gallery.html"))
val portrait = new Img

// gallery.html links to profile.html and two images
val gallery = new HTMLPage(Seq("kitten.jpg", "puppy.jpg"), Seq("profile.html"))
val kitten = new Img
val puppy = new Img

val internet = Map(
  "profile.html" -> profile,
  "gallery.html" -> gallery,
  "portrait.jpg" -> portrait,
  "kitten.jpg" -> kitten,
  "puppy.jpg" -> puppy
)

// fetch(url) attempts to fetch a resource from our fake internet.
// Its returned Future might contain a Resource or an exception
def fetch(url: String) = { new Promise(Try(internet(url))) }

順序組合

假設給定一個頁面URL，而你希望獲取該頁面的第一個圖。也許你正在做一個網站，在上面用戶可以發(fā)布有趣的網頁鏈接。為了幫助其他用戶決定某個鏈接是否值得追蹤，你打算顯示那個鏈接中第一張圖像的縮略圖。

即使你不知道組合子，你仍然可以寫一個縮略圖獲取函數：

def getThumbnail(url: String): Future[Resource]={
  val returnVal = new Promise[Resource]

  fetch(url) onSuccess { page => // callback for successful page fetch
    fetch(page.imageLinks()(0)) onSuccess { p => // callback for successful img fetch
      returnVal.setValue(p)
    } onFailure { exc => // callback for failed img fetch
      returnVal.setException(exc)
    }
  } onFailure { exc => // callback for failed page fetch
    returnVal.setException(exc)
  }
  returnVal
}

這個版本的函數能工作。它的大部分內容用來解析Future，然后把他們的內容傳給另一個Future。

我們希望得到一個頁面，然后從該頁面獲得一個圖像。如果你想獲得，然后再獲得的，這通常意味著順序組合。由于是異步的，所以需要使用flatMap：

def getThumbnail(url: String): Future[Resource] =
  fetch(url) flatMap { page => fetch(page.imageLinks()(0)) }

…通過并發(fā)組合

抓取頁面的第一個圖片是好的，但也許我們應該獲取所有圖片，并讓用戶自己進行選擇。我們可以使用for循環(huán)一個個地抓取，但這需要很長時間；所以我們想并行獲取它們。如果你想的事情“并行”發(fā)生，這通常意味著并發(fā)組合。所以我們使用Future.collect的提取所有的圖像：

def getThumbnails(url:String): Future[Seq[Resource]] =
  fetch(url) flatMap { page =>
    Future.collect(
      page.imageLinks map { u => fetch(u) }
    )
  }

如果這對你有意義，那太好了。你可能會看不懂這行代碼 page.imageLinks map { u => fetch(u) }：它使用map和map后的函數返回一個Future。當接下來的事情是返回一個Future時，我們不是應該使用flatMap嗎？但是請注意，在map前的不是一個Future；它是一個集合。map返回一個集合；我們使用Future.collect收集Future的集合到一個Future中。

并發(fā) + 遞歸

除了頁面中的圖片以外，我們可能會想獲取它鏈接的其他頁面。通過遞歸我們可以構建一個簡單的網絡爬蟲。

// Return
def crawl(url: String): Future[Seq[Resource]] =
  fetch(url) flatMap { page =>
    Future.collect(
      page.links map { u => crawl(u) }
    ) map { pps => pps.flatten }
}

crawl("profile.html")
   ...hangs REPL, infinite loop...
Ctrl-C
Execution interrupted by signal.


scala>
// She's gone rogue, captain! Have to take her out!
// Calling Thread.stop on runaway Thread[Thread-93,5,main] with offending code:
// scala> crawl("profile.html")

在實踐中，這個網絡爬蟲不是很有用：首先我們沒有告訴它何時停止爬行；其次即使資源剛剛被獲取過，它仍然會不厭其煩地重新獲取。

服務

一個Finagle 服務用來處理RPC，讀取請求并給予回復的。服務是針對請求和回應的一個函數Req => Future[Rep]。

abstract class Service[-Req, +Rep] extends (Req =>  Future[Rep])

Client and Server

在服務中，我們要同時定義客戶端和服務器。

一個Finagle客戶端“引入”一個網絡服務。從概念上講，Finagle客戶端由兩部分組成

  • 一個使用服務的函數：分發(fā)一個 Req并處理 Future[Rep]

  • 配置怎樣分發(fā)這些請求；例如，作為HTTP請求發(fā)送到api.twitter.com的80端口

同樣，Finagle服務端“輸出”網絡服務。一個服務端由兩個部分組成：

  • 一個實現服務的函數：傳入一個Req 并返回一個Future[Rep]

  • 配置如何“監(jiān)聽”輸入的 Reqs；例如，在80端口的HTTP請求。

這種設計分離了服務的“業(yè)務邏輯”和數據如何在網絡中流動的配置。

Filter and Server

我們也談論Finagle“過濾器”。過濾器在服務之間，修改流經它的數據。過濾器可以很好地和服務組合在一起。例如，如果你有一個速率限制過濾器和一個tweet服務，你可以把它們組合在一起形成有速率限制的tweet服務。

客戶端

一個Finagle客戶端“引入”一個網絡服務。它有一些配置來設定如何在網絡上發(fā)送數據。一個簡單的HTTP客戶端可能看起來像這樣：

import org.jboss.netty.handler.codec.http.{DefaultHttpRequest, HttpRequest, HttpResponse, HttpVersion, HttpMethod}
import com.twitter.finagle.Service
import com.twitter.finagle.builder.ClientBuilder
import com.twitter.finagle.http.Http

// Don't worry, we discuss this magic "ClientBuilder" later
val client: Service[HttpRequest, HttpResponse] = ClientBuilder()
  .codec(Http())
  .hosts("twitter.com:80") // If >1 host, client does simple load-balancing
  .hostConnectionLimit(1)
  .build()

val req = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/")

val f = client(req) // Client, send the request

// Handle the response:
f onSuccess { res =>
  println("got response", res)
} onFailure { exc =>
  println("failed :-(", exc)
}

服務端

一個服務端按服務進行定義，并配置如何“監(jiān)聽”網絡上的請求。一個簡單的HTTP服務端可能看起來像這樣：

import com.twitter.finagle.Service
import com.twitter.finagle.http.Http
import com.twitter.util.Future
import org.jboss.netty.handler.codec.http.{DefaultHttpResponse, HttpVersion, HttpResponseStatus, HttpRequest, HttpResponse}
import java.net.{SocketAddress, InetSocketAddress}
import com.twitter.finagle.builder.{Server, ServerBuilder}
import com.twitter.finagle.builder.ServerBuilder

// Define our service: OK response for root, 404 for other paths
val rootService = new Service[HttpRequest, HttpResponse] {
  def apply(request: HttpRequest) = {
    val r = request.getUri match {
      case "/" => new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK)
      case _ => new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND)
    }
    Future.value(r)
  }
}

// Serve our service on a port
val address: SocketAddress = new InetSocketAddress(10000)
val server: Server = ServerBuilder()
  .codec(Http())
  .bindTo(address)
  .name("HttpServer")
  .build(rootService)

這個`name`是我們強加的，雖然沒有在例子中使用它，但這個字段對分析和調試是很有用的。

過濾器

過濾器改造服務，它們可以提供通用的服務功能。例如你有幾個服務需要支持速率限制，這時可以寫一個限速過濾器并將其應用于所有的服務就解決問題了。過濾器也可以將服務分解成不同的階段。

一個簡單的代理可能看起來像這樣：

class MyService(client: Service[..]) extends Service[HttpRequest, HttpResponse]
{
  def apply(request: HttpRequest) = {
    client(rewriteReq(request)) map { res =>
      rewriteRes(res)
    }
  }
}

其中rewriteReq 和 rewriteRes可以提供協議翻譯，例如。

abstract class Filter[-ReqIn, +RepOut, +ReqOut, -RepIn]
  extends ((ReqIn, Service[ReqOut, RepIn]) => Future[RepOut])

通過圖示可以更清晰地看出其類型：

    ((ReqIn, Service[ReqOut, RepIn])
         => Future[RepOut])


          (*   Service   *)
[ReqIn -> (ReqOut -> RepIn) -> RepOut]

下面的例子展示了怎樣通過過濾器來提供服務超時機制。

class TimeoutFilter[Req, Rep](
  timeout: Duration,
  exception: RequestTimeoutException,
  timer: Timer)
  extends Filter[Req, Rep, Req, Rep]
{
  def this(timeout: Duration, timer: Timer) =
    this(timeout, new IndividualRequestTimeoutException(timeout), timer)

  def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = {
    val res = service(request)

    res.within(timer, timeout) rescue {
      case _: java.util.concurrent.TimeoutException =>
        res.cancel()
        Trace.record(TimeoutFilter.TimeoutAnnotation)
        Future.exception(exception)
    }
  }
}

這個例子展示了怎樣（通過認證服務）提供身份驗證來將 Service[AuthHttpReq, HttpRep] 轉換為 Service[HttpReq, HttpRep]。

class RequireAuthentication(authService: AuthService)
  extends Filter[HttpReq, HttpRep, AuthHttpReq, HttpRep] {
  def apply(
    req: HttpReq,
    service: Service[AuthHttpReq, HttpRep]
  ) = {
    authService.auth(req) flatMap {
      case AuthResult(AuthResultCode.OK, Some(passport), _) =>
        service(AuthHttpReq(req, passport))
      case ar: AuthResult =>
        Future.exception(
          new RequestUnauthenticated(ar.resultCode))
    }
  }
}

這樣使用過濾器是有好處的。它可以幫助你將“身份驗證邏輯”固定在一個地方。擁有一個獨立的類型執(zhí)行請求授權，會使追查程序安全問題變得更容易。

過濾器可以使用 andThen 組合在一起。傳入一個Service參數給andThen 將創(chuàng)建一個（添加了過濾功能）的Service（類型用來做說明）。

val authFilter: Filter[HttpReq, HttpRep, AuthHttpReq, HttpRep]
val timeoutfilter[Req, Rep]: Filter[Req, Rep, Req, Rep]
val serviceRequiringAuth: Service[AuthHttpReq, HttpRep]

val authenticateAndTimedOut: Filter[HttpReq, HttpRep, AuthHttpReq, HttpRep] =
  authFilter andThen timeoutFilter

val authenticatedTimedOutService: Service[HttpReq, HttpRep] =
  authenticateAndTimedOut andThen serviceRequiringAuth

生成器（Builder）

生成器把所有組件組合在一起。一個ClientBuilder對給定的一組參數生成一個Service，而一個 ServerBuilder 獲取一個 Service 的實例，并調度傳入請求給它。為了確定Service的類型，我們必須提供一個編解碼器（Codec）。編解碼器提供底層協議的實現（如HTTP，thrift，memcached）。這兩個Builder都有很多參數，其中一些是必填的。

下面是一個調用ClientBuilder的例子（類型用來做說明）

val client: Service[HttpRequest, HttpResponse] = ClientBuilder()
  .codec(Http)
  .hosts("host1.twitter.com:10000,host2.twitter.com:10001,host3.twitter.com:10003")
  .hostConnectionLimit(1)
  .tcpConnectTimeout(1.second)
  .retries(2)
  .reportTo(new OstrichStatsReceiver)
  .build()

這將構建一個客戶端在三個主機上進行負載平衡，最多在每臺主機建立一個連接，并在兩次失敗嘗試后放棄。統計數據會報給 ostrich 。以下生成器選項是必須的（而且它們也被靜態(tài)強制填寫了）：hosts 或 cluster, codec 和 hostConnectionLimit。

同樣的，你也可以使用一個ServerBuilder來創(chuàng)建“監(jiān)聽”傳入請求的服務：

val service = new MyService(...) // construct instance of your Finagle service
var filter = new MyFilter(...) // and maybe some filters
var filteredServce = filter andThen service
val  server = ServerBuilder()
  .bindTo(new InetSocketAddress(port))
  .codec(ThriftServerFramedCodec())
  .name("my filtered service")
//  .hostConnectionMaxLifeTime(5.minutes)
//  .readTimeout(2.minutes)
  .build(filteredService)

通過這些參數會生成一個Thrift服務器監(jiān)聽端口，并將請求分發(fā)給。如果我們去掉hostConnectionMaxLifeTime的注釋，每個連接將被允許留存長達5分鐘。如果我們去掉readTimeout的注釋，那么我們就需要在2分鐘之內發(fā)送請求。ServerBuilder必選項有：name, bindTo 和 codec。

不要阻塞（除非你用正確的方式）

Finagle 自動操縱線程來保證服務順利運行。但是，如果你的服務阻塞了，它會阻塞所有Finagle線程。

  • 如果你的代碼調用了一個阻塞操作（apply 或 get），使用Future  池來包裝阻塞代碼。阻塞操作將運行在自己的線程池中，返回一個Future來完成（或失?。┻@個操作，并可以和其它Future組合。

  • 如果你的代碼中使用Future的順序組合，不用擔心它會“阻塞”組合中的Future。

1 小心，還有其它“Future”類。不要將com.twitter.util.Future 和scala.actor.Future 或 java.util.concurrent.Future混淆起來！

2 如果你學習類型系統和/或分類理論，你會高興地發(fā)現flatMap相當于一元綁定。

Previous article： Next article：