?
Ce document utilise Manuel du site Web PHP chinois Libérer
Finagle-Friendly REPL
我們將要討論的不是標(biāo)準(zhǔn)Scala的代碼。如果你喜歡使用REPL學(xué)習(xí)，你可能想知道如何獲得一個(gè)加入Finagle 及其依賴的 Scala REPL。
你可以在這里獲取Finagle源代碼。
如果你在 finagle目錄下有Finagle的源代碼，你可以通過下面的命令得到一個(gè)控制臺(tái)
$ cd finagle $ ./sbt "project finagle-http" console ...build output... scala>
Futures
Finagle使用 com.twitter.util.Future1 編碼延遲操作。Future是尚未生成的值的一個(gè)句柄。Finagle使用Future作為其異步API的返回值。同步API會(huì)在返回前等待結(jié)果；但是異步API則不會(huì)等待。例如，個(gè)對(duì)互聯(lián)網(wǎng)上一些服務(wù)的HTTP請(qǐng)求可能半秒都不會(huì)返回。你不希望你的程序阻塞等待半秒。
“慢”的API可以立即返回一個(gè)Future，然后在需要解析其值時(shí)“填充”。
val myFuture = MySlowService(request) // returns right away ...do other things... val serviceResult = myFuture.get() // blocks until service "fills in" myFuture
在實(shí)踐中，你不會(huì)發(fā)送一個(gè)請(qǐng)求，然后在幾行代碼后調(diào)用myFuture.get。Future提供了注冊(cè)回調(diào)的方法，在值變得可用時(shí)會(huì)調(diào)用注冊(cè)的回調(diào)函數(shù)。
如果你用過其他異步API，當(dāng)看到“回調(diào)”你也許會(huì)畏縮。你可能會(huì)聯(lián)想到他們難以辨認(rèn)的代碼流，被調(diào)用的函數(shù)藏在離調(diào)用處遠(yuǎn)遠(yuǎn)的地方。但是，F(xiàn)uture可以利用Scala中“函數(shù)是一等公民”的特性編寫出更可讀的代碼流。你可以在調(diào)用它的地方簡(jiǎn)單地定義一個(gè)處理函數(shù)。
例如，寫代碼調(diào)度請(qǐng)求，然后“處理”回應(yīng)，你可以保持代碼在一起：
val future = dispatch(req) // returns immediately, but future is "empty"
future onSuccess { reply => // when the future gets "filled", use its value
println(reply)
}你可以在REPL中用體驗(yàn)一下Future。雖然不是學(xué)習(xí)如何在實(shí)際代碼中使用他們的好方法，但可以幫助理解API。當(dāng)你使用REPL，Promise是一個(gè)方便的類。它是Future抽象類的一個(gè)具體子類。你可以用它來創(chuàng)建一個(gè)還沒有值的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>在實(shí)際代碼中使用Future時(shí)，你通常不會(huì)調(diào)用get ，而是使用回調(diào)函數(shù)。
get僅僅是方便在REPL修修補(bǔ)補(bǔ)。
順序組合（Sequential composition）
Future有類似集合API中的組合子（如 map, flatMap） ?；仡櫼幌录辖M合子，它讓你可以表達(dá)如 “我有一個(gè)整數(shù)List和一個(gè)square函數(shù)：map那個(gè)列表獲得整數(shù)平方的列表”這樣的操作。這種表達(dá)方式很靈巧；你可以把組合子函數(shù)和另一個(gè)函數(shù)放在一起有效地組成一個(gè)新函數(shù)。面向Future的組合子可以讓你這樣表達(dá)：“我有一個(gè)期望整數(shù)的Future和一個(gè)square函數(shù)：map那個(gè)Future獲得一個(gè)期望整數(shù)平方的Future”。
如果你在定義一個(gè)異步API，傳入一個(gè)請(qǐng)求值，你的API應(yīng)該返回一個(gè)包裝在Future中的響應(yīng)。因此，這些把輸入和函數(shù)加入Future的組合子是相當(dāng)有用的：它們幫助你根據(jù)其它異步API定義你自己的異步API。
最重要的Future的組合子是flatMap2：
def Future[A].flatMap[B](f: A => Future[B]): Future[B]
flatMap 序列化兩個(gè)Future。即，它接受一個(gè)Future和一個(gè)異步函數(shù)，并返回另一個(gè)Future。方法簽名中是這樣寫的：給定一個(gè)Future成功的值，函數(shù)f提供下一個(gè)Future。如果/當(dāng)輸入的Future 成功完成，flatMap自動(dòng)調(diào)用f。只有當(dāng)這兩個(gè)Future都已完成，此操作所代表的Future才算完成。如果任何一個(gè)Future失敗，則操作確定的
Future也將失敗。這種隱交織的錯(cuò)誤讓我們只需要在必要時(shí)來處理錯(cuò)誤，所以語法意義很大。flatMap是這些語義組合子的標(biāo)準(zhǔn)名稱。
如果你有一個(gè)Future并且想在異步API使用其值，使用flatMap。例如，假設(shè)你有一個(gè)Future[User]，需要一個(gè)Future[Boolean]表示用戶是否已被禁止。有一個(gè)isBanned 的異步API來判斷一個(gè)用戶是否已被禁止。此時(shí)可以使用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中應(yīng)用一個(gè)同步函數(shù)，可以使用map。例如，假設(shè)你有一個(gè)Future[RawCredentials]需要一個(gè)Future[Credentials]。你有一個(gè)的同步的normalize函數(shù)將RawCredentials轉(zhuǎn)換成Credentials。可以使用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有快捷語法來調(diào)用flatMap：for表達(dá)式。假設(shè)你想通過異步API驗(yàn)證登錄請(qǐng)求，然后通過另一個(gè)異步API檢查用戶是否被禁止。在for表達(dá)式的幫助下，我們可以這樣寫：
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>它產(chǎn)生一個(gè)f: Future[(User,
Boolean)]，包含用戶對(duì)象和一個(gè)表示該用戶是否已被禁止的布爾值。注意這里是怎樣實(shí)現(xiàn)順序組合的：isBanned使用了authenticate的輸出作為其輸入。
并發(fā)組合
你可能想一次獲取來自多個(gè)服務(wù)的數(shù)據(jù)。例如，如果你正在編寫一個(gè)Web服務(wù)來顯示內(nèi)容和廣告，它可能會(huì)從兩個(gè)服務(wù)中分別獲取內(nèi)容和廣告。但是，你怎么告訴代碼來等待兩份答復(fù)呢？如果必須自己實(shí)現(xiàn)可能會(huì)非常棘手，幸運(yùn)的是你可以使用并發(fā)組合子。
Future 提供了一些并發(fā)組合子。一般來說，他們都是將Future的一個(gè)序列轉(zhuǎn)換成包含一個(gè)序列的Future，只是方式略微不同。這是很好的，因?yàn)樗ū举|(zhì)上）可以讓你把幾個(gè)Future封裝成一個(gè)單一的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參數(shù)是具有相同類型Future的一個(gè)集合，返回一個(gè)Future，其類型是包含那個(gè)類型值的一個(gè)序列。當(dāng)所有的Future都成功完成或者當(dāng)中任何一個(gè)失敗，都會(huì)使這個(gè)Future完成。返回序列的順序和傳入序列的順序相對(duì)應(yīng)。
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參數(shù)是混合類型的Future序列，返回一個(gè)Future[Unit]，當(dāng)所有的相關(guān)Future完成時(shí)（無論他們是否失?。┰揊uture完成。其作用是標(biāo)識(shí)一組異構(gòu)操作完成。對(duì)那個(gè)內(nèi)容和廣告的例子來說，這可能是一個(gè)很好的解決方案。
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>
當(dāng)傳入的Future序列的第一個(gè)Future完成的時(shí)候，select會(huì)返回一個(gè)Future。它會(huì)將那個(gè)完成的Future和其它未完成的Future一起放在Seq中返回。
（它不會(huì)做任何事情來取消剩余的Future。你可以等待更多的回應(yīng)，或者忽略他們）
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>
組合例子：緩存速率限制
這些組合子表達(dá)了典型的網(wǎng)絡(luò)服務(wù)操作。這段假設(shè)的代碼在對(duì)速率進(jìn)行限制（為了保持本地速率限制緩存）的同時(shí)，將用戶的請(qǐng)求調(diào)度到后臺(tái)服務(wù)：
// 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)
}
}
}這個(gè)例子結(jié)合了順序和并發(fā)組合。請(qǐng)注意，除了給轉(zhuǎn)化速率限制回應(yīng)一個(gè)異常以外，沒有明確的錯(cuò)誤處理。如果任何Future在這里失敗，它會(huì)自動(dòng)傳播到返回的Future中。
組合例子：網(wǎng)絡(luò)爬蟲
你已經(jīng)看到了怎樣使用Future組合子的例子，不過也許意猶未盡。假設(shè)你有一個(gè)簡(jiǎn)單的互聯(lián)網(wǎng)模型。該互聯(lián)網(wǎng)中只有HTML網(wǎng)頁(yè)和圖片，其中頁(yè)面可以鏈接到圖像和其他網(wǎng)頁(yè)。你可以獲取一個(gè)頁(yè)面或圖像，但API是異步的。這個(gè)假設(shè)的API成這些“可獲取”的數(shù)據(jù)為資源：
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))) }順序組合
假設(shè)給定一個(gè)頁(yè)面URL，而你希望獲取該頁(yè)面的第一個(gè)圖。也許你正在做一個(gè)網(wǎng)站，在上面用戶可以發(fā)布有趣的網(wǎng)頁(yè)鏈接。為了幫助其他用戶決定某個(gè)鏈接是否值得追蹤，你打算顯示那個(gè)鏈接中第一張圖像的縮略圖。
即使你不知道組合子，你仍然可以寫一個(gè)縮略圖獲取函數(shù)：
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
}這個(gè)版本的函數(shù)能工作。它的大部分內(nèi)容用來解析Future，然后把他們的內(nèi)容傳給另一個(gè)Future。
我們希望得到一個(gè)頁(yè)面，然后從該頁(yè)面獲得一個(gè)圖像。如果你想獲得，然后再獲得的，這通常意味著順序組合。由于是異步的，所以需要使用flatMap：
def getThumbnail(url: String): Future[Resource] =
fetch(url) flatMap { page => fetch(page.imageLinks()(0)) }…通過并發(fā)組合
抓取頁(yè)面的第一個(gè)圖片是好的，但也許我們應(yīng)該獲取所有圖片，并讓用戶自己進(jìn)行選擇。我們可以使用for循環(huán)一個(gè)個(gè)地抓取，但這需要很長(zhǎng)時(shí)間；所以我們想并行獲取它們。如果你想的事情“并行”發(fā)生，這通常意味著并發(fā)組合。所以我們使用Future.collect的提取所有的圖像：
def getThumbnails(url:String): Future[Seq[Resource]] =
fetch(url) flatMap { page =>
Future.collect(
page.imageLinks map { u => fetch(u) }
)
}如果這對(duì)你有意義，那太好了。你可能會(huì)看不懂這行代碼 page.imageLinks map { u => fetch(u)
}：它使用map和map后的函數(shù)返回一個(gè)Future。當(dāng)接下來的事情是返回一個(gè)Future時(shí)，我們不是應(yīng)該使用flatMap嗎？但是請(qǐng)注意，在map前的不是一個(gè)Future；它是一個(gè)集合。map返回一個(gè)集合；我們使用Future.collect收集Future的集合到一個(gè)Future中。
并發(fā) + 遞歸
除了頁(yè)面中的圖片以外，我們可能會(huì)想獲取它鏈接的其他頁(yè)面。通過遞歸我們可以構(gòu)建一個(gè)簡(jiǎn)單的網(wǎng)絡(luò)爬蟲。
// 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")在實(shí)踐中，這個(gè)網(wǎng)絡(luò)爬蟲不是很有用：首先我們沒有告訴它何時(shí)停止爬行；其次即使資源剛剛被獲取過，它仍然會(huì)不厭其煩地重新獲取。
服務(wù)
一個(gè)Finagle 服務(wù)用來處理RPC，讀取請(qǐng)求并給予回復(fù)的。服務(wù)是針對(duì)請(qǐng)求和回應(yīng)的一個(gè)函數(shù)Req =>
Future[Rep]。
abstract class Service[-Req, +Rep] extends (Req => Future[Rep])
在服務(wù)中，我們要同時(shí)定義客戶端和服務(wù)器。
一個(gè)Finagle客戶端“引入”一個(gè)網(wǎng)絡(luò)服務(wù)。從概念上講，F(xiàn)inagle客戶端由兩部分組成
一個(gè)使用服務(wù)的函數(shù)：分發(fā)一個(gè)
Req并處理Future[Rep]配置怎樣分發(fā)這些請(qǐng)求；例如，作為HTTP請(qǐng)求發(fā)送到
api.twitter.com的80端口
同樣，F(xiàn)inagle服務(wù)端“輸出”網(wǎng)絡(luò)服務(wù)。一個(gè)服務(wù)端由兩個(gè)部分組成：
一個(gè)實(shí)現(xiàn)服務(wù)的函數(shù)：傳入一個(gè)
Req并返回一個(gè)Future[Rep]配置如何“監(jiān)聽”輸入的 Reqs；例如，在80端口的HTTP請(qǐng)求。
這種設(shè)計(jì)分離了服務(wù)的“業(yè)務(wù)邏輯”和數(shù)據(jù)如何在網(wǎng)絡(luò)中流動(dòng)的配置。
我們也談?wù)揊inagle“過濾器”。過濾器在服務(wù)之間，修改流經(jīng)它的數(shù)據(jù)。過濾器可以很好地和服務(wù)組合在一起。例如，如果你有一個(gè)速率限制過濾器和一個(gè)tweet服務(wù)，你可以把它們組合在一起形成有速率限制的tweet服務(wù)。
客戶端
一個(gè)Finagle客戶端“引入”一個(gè)網(wǎng)絡(luò)服務(wù)。它有一些配置來設(shè)定如何在網(wǎng)絡(luò)上發(fā)送數(shù)據(jù)。一個(gè)簡(jiǎn)單的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)
}服務(wù)端
一個(gè)服務(wù)端按服務(wù)進(jìn)行定義，并配置如何“監(jiān)聽”網(wǎng)絡(luò)上的請(qǐng)求。一個(gè)簡(jiǎn)單的HTTP服務(wù)端可能看起來像這樣：
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)這個(gè)`name`是我們強(qiáng)加的，雖然沒有在例子中使用它，但這個(gè)字段對(duì)分析和調(diào)試是很有用的。
過濾器
過濾器改造服務(wù)，它們可以提供通用的服務(wù)功能。例如你有幾個(gè)服務(wù)需要支持速率限制，這時(shí)可以寫一個(gè)限速過濾器并將其應(yīng)用于所有的服務(wù)就解決問題了。過濾器也可以將服務(wù)分解成不同的階段。
一個(gè)簡(jiǎn)單的代理可能看起來像這樣：
class MyService(client: Service[..]) extends Service[HttpRequest, HttpResponse]
{
def apply(request: HttpRequest) = {
client(rewriteReq(request)) map { res =>
rewriteRes(res)
}
}
}其中rewriteReq 和 rewriteRes可以提供協(xié)議翻譯，例如。
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]
下面的例子展示了怎樣通過過濾器來提供服務(wù)超時(shí)機(jī)制。
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)
}
}
}這個(gè)例子展示了怎樣（通過認(rèn)證服務(wù)）提供身份驗(yàn)證來將 Service[AuthHttpReq, HttpRep] 轉(zhuǎn)換為
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))
}
}
}這樣使用過濾器是有好處的。它可以幫助你將“身份驗(yàn)證邏輯”固定在一個(gè)地方。擁有一個(gè)獨(dú)立的類型執(zhí)行請(qǐng)求授權(quán)，會(huì)使追查程序安全問題變得更容易。
過濾器可以使用 andThen 組合在一起。傳入一個(gè)Service參數(shù)給andThen 將創(chuàng)建一個(gè)（添加了過濾功能）的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）
生成器把所有組件組合在一起。一個(gè)ClientBuilder對(duì)給定的一組參數(shù)生成一個(gè)Service，而一個(gè)
ServerBuilder 獲取一個(gè) Service 的實(shí)例，并調(diào)度傳入請(qǐng)求給它。為了確定Service的類型，我們必須提供一個(gè)編解碼器（Codec）。編解碼器提供底層協(xié)議的實(shí)現(xiàn)（如HTTP，thrift，memcached）。這兩個(gè)Builder都有很多參數(shù)，其中一些是必填的。
下面是一個(gè)調(diào)用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()這將構(gòu)建一個(gè)客戶端在三個(gè)主機(jī)上進(jìn)行負(fù)載平衡，最多在每臺(tái)主機(jī)建立一個(gè)連接，并在兩次失敗嘗試后放棄。統(tǒng)計(jì)數(shù)據(jù)會(huì)報(bào)給 ostrich 。以下生成器選項(xiàng)是必須的（而且它們也被靜態(tài)強(qiáng)制填寫了）：hosts 或 cluster,
codec 和 hostConnectionLimit。
同樣的，你也可以使用一個(gè)ServerBuilder來創(chuàng)建“監(jiān)聽”傳入請(qǐng)求的服務(wù)：
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)通過這些參數(shù)會(huì)生成一個(gè)Thrift服務(wù)器監(jiān)聽端口，并將請(qǐng)求分發(fā)給。如果我們?nèi)サ?code>hostConnectionMaxLifeTime的注釋，每個(gè)連接將被允許留存長(zhǎng)達(dá)5分鐘。如果我們?nèi)サ?code>readTimeout的注釋，那么我們就需要在2分鐘之內(nèi)發(fā)送請(qǐng)求。ServerBuilder必選項(xiàng)有：name,
bindTo 和 codec。
不要阻塞（除非你用正確的方式）
Finagle 自動(dòng)操縱線程來保證服務(wù)順利運(yùn)行。但是，如果你的服務(wù)阻塞了，它會(huì)阻塞所有Finagle線程。
如果你的代碼調(diào)用了一個(gè)阻塞操作（
apply或get），使用Future 池來包裝阻塞代碼。阻塞操作將運(yùn)行在自己的線程池中，返回一個(gè)Future來完成（或失敗）這個(gè)操作，并可以和其它Future組合。
如果你的代碼中使用Future的順序組合，不用擔(dān)心它會(huì)“阻塞”組合中的Future。
1 小心，還有其它“Future”類。不要將com.twitter.util.Future 和scala.actor.Future 或
java.util.concurrent.Future混淆起來！
2 如果你學(xué)習(xí)類型系統(tǒng)和/或分類理論，你會(huì)高興地發(fā)現(xiàn)flatMap相當(dāng)于一元綁定。