细说Kestrel.scala中的PersistentQueue

本文是Scala代码实例之Kestrel的第五部分，继续讲述PersistentQueue处理消息队列并发请求的方式。

成都创新互联公司是一家集网站建设,景宁畲族自治企业网站建设,景宁畲族自治品牌网站建设,网站定制,景宁畲族自治网站建设报价,网络营销,网络优化,景宁畲族自治网站推广为一体的创新建站企业，帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿，时刻以成就客户成长自我，坚持不断学习、思考、沉淀、净化自己，让我们为更多的企业打造出实用型网站。

回顾一下之前我们读过的两个文件，Kestrel.scala, QueueCollection.scala。Kestrel.scala是启动文件，并且通过一个actor，保持整个项目不会因为没有线程运行而退出，同时注册了一个acceptor，当建立起新的链接的时候，访问 KestrelHandler.scala(这个稍后我们再读)。QueueCollection.scala，维护一个PersistentQueue的队列，如果访问的queue_name不存在，则创建一个，如果存在，就对相应的QueueCollection进行操作。如果留心的话，我们还可以看到QueueCollection在启动的时候，queue_name的来源是一个文件目录。

我们就从这个入口继续往下，看看PersistentQueue是如何处理消息队列的并发请求的：

在前几篇文章里面，我们曾经提到过PersistentQueue有两个“类”，一个是object PersistentQueue，一个是class PersistentQueue。而object在scala是一个单例模式，也就是singleton。也可以看做是只有static类型的java类。现在让我们关注一下，看看class PersistentQueue和object Persistent之间的关系是怎样的。

刚开始的一段代码有点吓人：

 
 
 
    
  
  
  
  1. class OverlaySetting[T](base: => T) {  
    2. 
  
  
  
  2.   @volatile private var local: Option[T] = None  
    3. 
  
  
  
  3.   def set(value: Option[T]) = local = value  
    4. 
  
  
  
  4.   def apply() = local.getOrElse(base)  
    5. 
  
  
  
  5. }  
    6.

我们先跳过去，直接往下看，看到这里：

 
 
 
    
  
  
  
  1. def overlay[T](base: => T) = new OverlaySetting(base)  
    2. 
  
  
  
  2. // attempting to add an item after the queue reaches this size (in items) will fail.  
    3. 
  
  
  
  3. val maxItems = overlay(PersistentQueue.maxItems)  
    4. 
  
  
  
  4. // attempting to add an item after the queue reaches this size (in bytes) will fail.  
    5. 
  
  
  
  5. val maxSize = overlay(PersistentQueue.maxSize)  
    6. 
  
  
  
  6. ……  
    7.

如果我们不细究overlay的内容，这段代码其实就是把object PersisitentQueue中的变量赋值给class PersistentQueue中，那么overlay究竟做了什么呢？其实，overlay是将变量做了一个封装，封装在一个叫做OverlaySetting的类里面。这个类，根据我们之前对scala语法的了解，可以知道，它是一个OverlaySetting[T]的类，并且在创建的时候，需要带入方法，方法没有参数，但是有一个返回值，类型就是T。(关于class类的语法规则，可以参考http://programming-scala.labs.oreilly.com/ch05.html#Constructors，不过里面的例子比OverlaySetting还复杂……-_-|||)

这个类在每次创建对象的时候，都会被赋值。我们也看到只有在使用apply方法的时候才会被调用（不过我没有太想明白，如何通过函数的返回值来确定模板中的类型T，也许这就是Scala这种更加灵活的编译算法，可以在new对象的时候，通过审查变量类型来获取T的吧，毕竟Scala是一个静态语言，如果是动态语言就不太成为一个问题了）。

这里面还存在一个Scala概念，就是方法＝变量。当然在很多动态语言里面就已经这么做了。在Scala里面，我们可以把def看作是val的一种特殊写法，def声明的方法，也可以用 def func_name() = {} 这样的语法规则，跟val基本就是一回事了。当然，这一改变在Scala里面并不简单是一个语法规则的问题，更进一步的，所有的变量也都是类，所以我们可以把一个变量，看做一个类，也可以看做类的建构函数，返回的就是类本身……有点绕，不过这样理解，就比较好理解为什么可以用常量，当作没有参数的方法调用了。

说了那么多，结论很简单，maxSize是一个OverlaySetting[LONG]的类，如果maxSize没有设置过，那么返回的就是object PersistentQueue里面的maxSize。LONG类型。

在主程序体里面，我们看到了Journal类，然后是调用 configure 方法，这个方法印证了我们的对OverlaySetting的解释，它从配置文件里面把参数都读出来赋值给class PersistentQueue里面的那些常量，用的是set。这里是一个Scala的语法细节，它省略了一些不必要的”.”和”()”。

休息一下。我们开始讨论在PersistentQueue里面的Actor

……

休息完毕

Scala中，消息传递的方式有一个特殊的语法结构：“Object ! MessageType” 就好像在源代码里面出现的：“w.actor ! ItemArrived。”，（关于Scala的Actor，详细的语法说明在http://programming-scala.labs.oreilly.com/ch09.html可以看到，建议先看一下，好对actor有一个比较深入的了解）

我们发现PersistentQueue中Actor的实现，跟语法说明里面的很不一样，在语法说明里面的Actor都是作为一个独立的线程出现的，而在PersistentQueue中，你甚至看不见一个对Actor的重载，但我们可以发现与Actor相关的几个地方，一个是Waiter的定义，它是一个case class，并且有一个成员变量叫做actor，类型是Actor：

 
 
 
    
  
  
  
  1. private case class Waiter(actor: Actor)  
    2. 
  
  
  
  2. ……  
    3. 
  
  
  
  3. private val waiters = new mutable.ArrayBuffer[Waiter]  
    4. 
  
  
  
  4. ……  
    5. 
  
  
  
  5.     val w = Waiter(Actor.self)  
    6. 
  
  
  
  6.     waiters += w  
    7. 
  
  
  
  7. ……  
    8.

需要注意：之前我们提过一个Scala的语法规则，那就是类后面的建构函数的参数，就是类中的成员变量！（不过这是在解释，为什么在建构函数里面会有private关键字时提到的……）所以，我们知道了一点，就是每一个Waiter内部都有一个actor，这些actor通过Actor.self共享了一个线程，当然也和其他的PersistentQueue共享了一个Actor。这是有点让人不习惯，因为这么要紧的一个线程的创建，竟然可以出现得那么隐蔽。甚至连一个大括号都没有。

接下来，我们来看看Actor是怎么在PersistentQueue里面工作了——这有点难，因为它的机制有点复杂，不是简单的象语法说明里面的那样，是一个完整的独立的函数，而是在一些函数中，突然切入进来，分享了Actor.self的一部分线程资源，就像下面代码一样：

 
 
 
    
  
  
  
  1. ……  
    2. 
  
  
  
  2. f operateReact(op: => Option[QItem], timeoutAbsolute: Long)(f: Option[QItem] => Unit): Unit = {  
    3. 
  
  
  
  3. operateOrWait(op, timeoutAbsolute) match {  
    4. 
  
  
  
  4.   case (item, None) =>  
    5. 
  
  
  
  5.     f(item)  
    6. 
  
  
  
  6.   case (None, Some(w)) =>  
    7. 
  
  
  
  7.     Actor.self.reactWithin((timeoutAbsolute - Time.now) max 0) {  
    8. 
  
  
  
  8.       case ItemArrived => operateReact(op, timeoutAbsolute)(f)  
    9. 
  
  
  
  9.       case TIMEOUT => synchronized {  
    10. 
  
  
  
  10.         waiters -= w  
    11. 
  
  
  
  11.         // race: someone could have done an add() between the timeout and grabbing the lock.  
    12. 
  
  
  
  12.         Actor.self.reactWithin(0) {  
    13. 
  
  
  
  13.           case ItemArrived => f(op)  
    14. 
  
  
  
  14.           case TIMEOUT => f(op)  
    15. 
  
  
  
  15.         }  
    16. 
  
  
  
  16.       }  
    17. 
  
  
  
  17.     }  
    18. 
  
  
  
  18.   case _ => throw new RuntimeException()  
    19. 
  
  
  
  19. }  
    20. 
  
  
  
  20.  
    21. 
  
  
  
  21. ……  
    22.

其中：

 
 
 
    
  
  
  
  1. Actor.self.reactWithin(0) {  
    2. 
  
  
  
  2.     case ItemArrived => f(op)  
    3. 
  
  
  
  3.     case TIMEOUT => f(op)  
    4. 
  
  
  
  4. }  
    5.

就是Actor的一个语法，在一段时间里面等待消息，如果有消息就如何……，如果没有消息（TIMEOUT），就如何……。但是在整个函数里面套用了两层 Actor.self.reactWithin，有点让人要晕菜的感觉，再加上之前有一个match…case的结构，调用了operateOrWait(op, timeoutAbsolute)方法。要了解整个消息处理的机制，就需要把这三个部分联系起来看了。

先简单看一下operateOrWait函数，比较容易理解：

 
 
 
    
  
  
  
  1. private def operateOrWait(op: => Option[QItem], timeoutAbsolute: Long): (Option[QItem], Option[Waiter]) = synchronized {  
    2. 
  
  
  
  2.   val item = op  
    3. 
  
  
  
  3.   if (!item.isDefined && !closed && !paused && timeoutAbsolute > 0) {  
    4. 
  
  
  
  4.     val w = Waiter(Actor.self)  
    5. 
  
  
  
  5.     waiters += w  
    6. 
  
  
  
  6.     (None, Some(w))  
    7. 
  
  
  
  7.   } else {  
    8. 
  
  
  
  8.     (item, None)  
    9. 
  
  
  
  9.   }  
    10. 
  
  
  
  10. }  
    11.

返回值是一个map，包括两个被Option封装的类型QItem和Waiter，从QItem.scala中可以知道（代码很简单），QItem就是把原始数据打了一个包，而Waiter之前我们也已经说过了。程序体中的判断是这样的：如果item，也就是op这个参数没有定义，并且PersistentQueue也没有停止，关闭，而且处理时间AbsoluteTime不是0，那么就创建一个Waiter，返回(None, Some[Waiter])；如果不满足这些条件，那么就直接返回(op, None)。简单的说，就是如果系统还能等，就让他等待正常一段时间然后操作，如果不能等，就直接返回操作指令。返回值只有两种类型。

然后再看operateReact，如果返回的是时间参数是None（详细的可以参考 actor .. case 的语法，地址是：http://programming-scala.labs.oreilly.com/ch03.html#MatchingOnCaseClasses)，那么就直接执行f(op)的函数，把op这个方法，作为参数传递给f函数。如果返回的是一个时间戳，Some(w)，那么我们就等待AbsoluteTime 到 Time.now()这段时间，如果在这段事件里面有ItemArrived事件发生，那么就处理一下，直到Time.now 等于或者大于 AbsoluteTime，那就会得到一个TIMEOUT，然后就退出了。（有一个异常的情况，需要清空一下事件队列，通过reactWithin(0){}）

这么理解这段actor还是不太清晰，那么让我们回到上一层的调用。看看这个f(op)到底是什么，然后我们看到了：

 
 
 
    
  
  
  
  1. def removeReact(timeoutAbsolute: Long, transaction: Boolean)(f: Option[QItem] => Unit): Unit = {  
    2. 
  
  
  
  2.   operateReact(remove(transaction), timeoutAbsolute)(f)  
    3. 
  
  
  
  3. }  
    4.

我们就知道op其实是一个remove的操作，并且返回remove得到的QItem对象。再往上一层到QueueCollection，我们看到：

 
 
 
    
  
  
  
  1. q.removeReact(if (timeout == 0) timeout else Time.now + timeout, transaction) {  
    2. 
  
  
  
  2.   case None =>  
    3. 
  
  
  
  3.     queueMisses.incr  
    4. 
  
  
  
  4.     f(None)  
    5. 
  
  
  
  5.   case Some(item) =>  
    6. 
  
  
  
  6.     queueHits.incr  
    7. 
  
  
  
  7.     f(Some(item))  
    8. 
  
  
  
  8. }  
    9.

f方法的操作，如果之前的remove返回的是一个None，则记录queueMess(未命中)添加1，如果返回的是一个QItem的值，那么就记录queueHits(命中)添加1，并且，对这个QItem进行操作（注意：这里的f是QueueCollection中remove带入的那个方法，而不是前面提到的removeReact里面提到的f。

从QueueCollection的remove调用到***层PersistentQueue的operateReact调用，我们大致可以了解这么曲折的调用关系解决了一个什么问题——从消息队列里面获取QItem。

回顾一下QueueCollection其他的代码，我们发现，只有waiter.size > 0的时候，有新的QItem添加，才会发出ItemArrived事件。也就是说，只有有一个获取消息队列的进程存在的时候，才会触发ItemArrived事件。获取消息队列，则通过使用reactWithin，允许在一个规定的时间内，连续处理一系列的ItemArrived事件。看QueueCollection的remove方法，我们还可以知道，当启动q.removeReact之前，首先会调用q.peek来检查，队列是不是为空，如果不是空的话，就直接返回队列里面最前面的那个元素。所以我们可以把这个消息队列理解成——如果消息队列为空的情况下，让获取消息队列的Client等待一段时间的机制，以降低反复进行SOCKET连接带来的不必要的耗损。

这个机制，可以让我们比较好地理解，为什么Kestrel提示说，如果运行多个独立的进程来处理消息队列的时候，会让这个消息队列的处理变成一个缺乏时序，但是处理并发能力很强的集群。每个连接对应的是一个Waiter，但是当ItemArrived触发的时候，只可能有其中的一个reactWithin得到了这个事件，发送给对应的那个线程处理这个消息。

我现在手上的是Kestrel-1.1.2版本的代码，走读这部分代码的时候，其实发现作者在写这段代码的时候，多了一些冗余的内容——比如说removeReceive方法，从而看出作者在使用Scala的特性中，也是逐步地把代码优化成如今的样子。毕竟Scala和Java之间的差别很大，如果做到Type Less, Do More。是需要一个逐步积累的过程，谁都不是天生就能把Scala写得很好的，更何况是需要性能非常高的时候。

【相关阅读】

  1. 从Java走进Scala（Scala经典读物）
  2. A Scala Tutorial for Java programmers
  3. 专题：Scala编程语言
  4. 从Scala看canEqual与正确的的equals实现
  5. Scala快速入门：从下载安装到定义方法

本文题目：细说Kestrel.scala中的PersistentQueue
转载来源：http://www.shufengxianlan.com/qtweb/news5/82905.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

解决方案知识

同城分类信息

网站建设方案　　 　成都网站设计　　 　四川德阳主机托管　　 　阿坝网站建设　　 　成都企业网站改版　　 　企业微信定制　　 　简阳机房托管　　 　LED显示屏　　 　wap网站制作　　 　犀浦防护栏定制　　 　成都做网站建设公司　　 　成都企业网站建设公司　　 　成都画册快印　　 　域名注册　　 　双流做网站　　 　化妆品企业网站建设方案　　 　力比特机械　　 　达州托管服务器　　 　网站运营　　 　乐山服务器托管