Searchbird

lue: String) = { log.debug("put %s", key) forward(key) = value // serialize updaters synchronized { value.split(" ").toSet foreach { token => val current = reverse.getOrElse(token, Set()) reverse(token) = current + key } } Future.Unit}需要注意的是（即使 HashMap 是线程安全的）同时只能有一个线程可以更新倒排索引，以确保对映射条目的 读-修改-写 是一个原子操作。 （这段代码过于保守；在进行 检索-修改-写 操作时，它锁定了整个映射，而不是锁定单个条目。）。另外还要注意使用 Set 作为数据结构；这可以确保即使一个文件中两次出现同样的符号，它也只会被 foreach 循环处理一次。这个实现仍然有一个问题，作为留给读者的一个练习：当我们用一个新文档覆盖的一个键的时候，我们诶有删除任何倒排索引中引用的旧文件。现在进入搜索引擎的核心：新的 search 方法。他应该解析查询，寻找匹配的文档，然后对这些列表做相交操作。这将产生包含所有查询中的标记的文件列表。在 Scala 中可以很直接地表达；添加这段代码到SearchbirdServiceImpl 类中：def search(query: String) = Future.value { val tokens = query.split(" ") val hits = tokens map { token => reverse.getOrElse(token, Set()) } val intersected = hits reduceLeftOption { _ & _ } getOrElse Set() intersected.toList}在这段短短的代码中有几件事情是值得关注的。在构建命中列表时，如果键（ token ）没有被发现， getOrElse 会返回其第二个参数（在这种情况下，一个空 Set ）。我们使用 left-reduce 执行实际的相交操作。特别是当 reduceLeftOption 发现 hits 为空时将不会继续尝试执行 reduce 操作。这使我们能够提供一个默认值，而不是抛出一个异常。其实这相当于：def search(query: String) = Future.value { val tokens = query.split(" ") val hits = tokens map { token => reverse.getOrElse(token, Set()) } if (hits.isEmpty) Nil else hits reduceLeft { _ & _ } toList}使用哪种方式大多是个人喜好的问题，虽然函数式风格往往会避开带有合理默认值的条件语句。现在，我们可以尝试在控制台中实验我们新的实现。重启服务器：$ ./sbt...> compile> run -f config/development.scala...[info] Running com.twitter.searchbird.Main -f config/development.scala然后再从 searchbird 目录，启动客户端：$ ./console 127.0.0.1 9999...[info] Running com.twitter.searchbird.SearchbirdConsoleClient 127.0.0.1 9999'client' is bound to your thrift client.finagle-client> 粘贴以下说明到控制台：client.put("basics", " values functions classes methods inheritance try catch finally expression oriented")client.put("basics", " case classes objects packages apply update functions are objects (uniform access principle) pattern")client.put("collections", " lists maps functional combinators (map foreach filter zip")client.put("pattern", " more functions! partialfunctions more pattern")client.put("type", " basic types and type polymorphism type inference variance bounds")client.put("advanced", " advanced types view bounds higher kinded types recursive types structural")client.put("simple", " all about sbt the standard scala build")client.put("more", " tour of the scala collections")client.put("testing", " write tests with specs a bdd testing framework for")client.put("concurrency", " runnable callable threads futures twitter")client.put("java", " java interop using scala from")client.put("searchbird", " building a distributed search engine using")现在，我们可以执行一些搜索，返回包含搜索词的文件的键。> client.search("functions").get()res12: Seq[String] = ArrayBuffer(basics)> client.search("java").get()res13: Seq[String] = ArrayBuffer(java)> client.search("java scala").get()res14: Seq[String] = ArrayBuffer(java)> client.search("functional").get()res15: Seq[String] = ArrayBuffer(collections)> client.search("sbt").get()res16: Seq[String] = ArrayBuffer(simple)> client.search("types").get()res17: Seq[String] = ArrayBuffer(type, advanced)回想一下，如果调用返回一个 Future ，我们必须使用一个阻塞的 get() 来获取其中包含的值。我们可以使用 Future.collect 命令来创建多个并发请求，并等待所有请求成功返回：> import com.twitter.util.Future...> Future.collect(Seq( client.search("types"), client.search("sbt"), client.search("functional") )).get()res18: Seq[Seq[String]] = ArrayBuffer(ArrayBuffer(type, advanced), ArrayBuffer(simple), ArrayBuffer(collections))分发我们的服务单台机器上一个简单的内存搜索引擎将无法搜索超过内存大小的语料库。现在，我们要大胆改进，用一个简单的分片计划来构建分布式节点。下面是框图：抽象为了帮助我们的工作，我们会先介绍另一个抽象索引来解耦 SearchbirdService 对索引实现的依赖。这是一个直观的重构。我们首先添加一个索引文件到构建 (创建文件 searchbird/src/main/scala/com/twitter/searchbird/Index.scala ):…/Index.scalapackage com.twitter.searchbirdimport scala.collection.mutableimport com.twitter.util._import com.twitter.conversions.time._import com.twitter.logging.Loggerimport com.twitter.finagle.builder.ClientBuilderimport com.twitter.finagle.thrift.ThriftClientFramedCodectrait Index { def get(key: String): Future[String] def put(key: String, value: String): Future[Unit] def search(key: String): Future[List[String]]}class ResidentIndex extends Index { val log = Logger.get(getClass) val forward = new mutable.HashMap[String, String] with mutable.SynchronizedMap[String, String] val r