一个简单的自定义配置格式的Scala实现

本来想执行我Object As Configuration的理念， 不过google group上有人推荐我看看configgy， 看过之后， 觉得还是自己写一个简单些， So 马上动手...

配置格式先简单定义为：

ident = value

groupId = {

ident1 = value1

ident2 = value2

...

}

即可以配置为key-value的形式（以=或者:分割），或者group的形式（group暂时不考虑嵌套group的形式）。

定义一个Scala的Parser combinator如下：

class BlockOrItemConfigurationParser extends JavaTokenParsers {

  def entry = ((item | block) +) ^^ {

    _.foldLeft(Map[String, Any]()) {

      (accum, i) =>

        i._2 match {

          case lst: List[(String, Any)] => accum ++ lst.foldLeft(Map[String, Any]()) {

            (ac, it) =>

              ac + ((i._1 + "." + it._1) -> it._2)

          }

          case x => accum + (i._1 -> x)

        }

    }

  }

  def block = ident ~ "=" ~ "{" ~ rep(item) ~ "}" ^^ {

    case k ~ "=" ~ "{" ~ v ~ "}" => (k -> v)

  }

  def item = ident ~ ("=" | ":") ~ value <~ opt(";") ^^ {

    case k ~ _ ~ v => (k -> v)

  }

  def value = stringLit | decimalNumber | floatingPointNumber | booleanLiteral | nullLiteral

  def stringLit = "\"" ~> string <~ "\"" ^^ {

    case s => s

  }

  def string = ("""([^"\p{Cntrl}\\]|\\[\\/bfnrt]|\\u[a-fA-F0-9]{4})*""").r

  def booleanLiteral = "true" ^^^ true | "false" ^^^ false

  def nullLiteral = "null" ^^^ null

}

代码写的有点儿矬， 将就看吧，呵呵， 本来用的JavaTokenParsers.stringLiteral， 但后来发现他没有去掉引号", 所以，只能自己重新定义一个， 当然，代码直接拷贝它的。

有了parser之后， 就可以定义一个Configurator，比如：

class SimpleConfigurator(configFile:File) {

   val p = new BlockOrItemConfigurationParser

   p.parseAll(p.entry, new CharSeqReader(FileUtils.readFileToString(configFile))) match{

case p.Success(r, in)=> // use r to get parsed result and feed your program, hehe 

case x=> throw new Exception("parse error:"+x)

}

}

OK， That's it！Have Fun!

Simple Introduction On Akka Actor

Simple Introduction On Akka Actor

王福强(Darren.Wang)

<fujohnwang@gmail.com>

---

Table of Contents

1. Actor Vs. Task

2. Akka Actor Quick Start

2.1. UnTyped Actor

2.1.1. Akka Actor的创建和启动

2.1.2. Akka Actor的使用

2.1.3. Stop or KillTheActor

2.1.4. Lifecycle Callbacks

2.1.5. Other Features

2.2. Typed Actor

3. Dispatcher

4. Fault Tolerance Facility

5. Remote Actor

6. Conclusion

Simple Introduction就是简单的介绍， 呵呵，所以， 这篇文字只是初级的浮于表面的泛泛之谈， 不要期望写的有多么事无巨细， 写得有多么深入人心，简单了解一下吧，感兴趣自己再找资料啃呗！

Akka Framework有一堆modules， 这里只扯它的Actor支持， 什么不知道Actor是啥？哦， 那先去看看Erlanga啦， Wikipedia啦， Scala Actor啦， 然后再回来看这篇文字，呵呵，因为我不会那么详细的告诉你Actor到底是个什么东西， ：）

1. Actor Vs. Task

Actor和Task那， 可以说是Share Nothing/Message Passing Concurrency跟Shared State Concurrency的标志性 “建筑物” ， 要比较很细的话， 可能也是一家有一家的话要说， 我那只说一点我的观点， 各位看官要是看得不爽， 也别喷， 欢迎纠正，使得我也可以迷途知返嘛！

我的观点，简单来说， Actor可以维护多次逻辑执行的状态， 而Task则不维护这种逻辑多次执行的状态。 举例来说， 你把一个Callable或者Runnable提交给ExecutorService，并且当这个Task(Callable和Runnable算是Task的某些具体形式)执行完成后， 它的状态就此完结， 不会对下一个提交并执行的Task产生任何影响（当然， task与task之间通过某种全局的状态管理的情况除外）； 而Actor则不同， 一笔按照Actor的逻辑执行后， 大部分会对下一笔程序同样逻辑执行产生影响，因为它们的执行状态是通过Actor本身来维护的。 本质上，这种情况算是基于Task的并发建模与基于数据的并发建模的差异所造成的。

扯完这些，我们下面看本篇文字的主角-Akka Actor。

2. Akka Actor Quick Start

Akka Actor是Scala平台上的一种Actor实现， 其它的还有Scala本身类库提供的默认Actor实现，Lift框架的Actor实现，等等， 但Akka相对于其它Actor实现来说， “忽悠” 的名声有点儿大（ 因为基本上所有Actor实现上， 一些基本的组件都会有的，只不过，它忽悠的也还好， 性能确实比Scala默认提供的Actor实现要好），所以这也就是为啥要简单介绍一下它的原因啦～

AkkaActor分两种， 一种叫Untyped Actor， 这种Actor就是常见的跟Erlang的ActorAPI实现和使用方式类似的那种， 而另一种叫Typed Actor, 它是通过API的强类型约束来声明和实现Actor， 后面会详细介绍它，我们先看Untyped Actor...

2.1. UnTyped Actor

Untyped Actor的实现很简单，只要实现akka.actor.Actor这个trait就行：

class MyActor extends Actor{
     protected def receive = {
       case msg:String=> println("receives message:"+msg)
       case _=> println("unexpected message")
     }
}
    

一个Untyped Actor，只要实现receive方法就行了， Akka会有外围的调度框架来调度和执行它。知道了怎么定义一个Akka Actor， 那下面就是怎么创建和使用它了。

2.1.1. Akka Actor的创建和启动

AkkaActor并不是直接实例化你的Actor定义并使用它，而是通过ActorRef来引用和管理你的Actor实例， 代码说话：

val myActor:ActorRef = actorOf[MyActor]
myActor.start     
     

akka.actor.Actor.actorOf工厂方法会根据你的Actor定义的类型来实例化和启动相应的Actor实例， 返回的Actor实例是ActorRef类型，而不是Actor类型，这个与Scala的Actor是不同的地方。（以上myActor定义的时候声明的类型是为了强调， 实际上因为有Scala的类型推导，此处的类型是可以省略的）

akka.actor.Actor.actorOf还有许多变体（重载）， 比如， 如果我们的Actor定义有构造方法参数：

class MyActor(id:String) extends Actor{
     protected def receive = {
       case msg:String=> println("receives message:"+msg)
       case _=> println("unexpected message")
     }
}
    

那么， 可以使用以下形式来实例化和启动Actor:

val myActor:ActorRef = actorOf(new Actor("my actor's id"))
myActor.start     
     

更多构造方式可以参阅Akka的Scaladoc， 此处有在线版本: http://akka.io/api/1.0/akka-actor/

	Tip
	Actor的实例化和启动也可以以chaining的方式一次搞定: val myActor:ActorRef = actorOf(new Actor("my actor's id")).start

2.1.2. Akka Actor的使用

Actor是以消息传递(Message Passing)的方式来实现并发的，所以， 使用Actor基本上就可以简化为Actor间的发送消息， 与大部分Actor一样， AkkaActor提供了以下几种消息发送方式:

1. ! - 异步消息发送.  !在这里读“bang”， 单一的!表示fire-and-forget形式的异步消息发送，该方法会即可返回而不用等消息处理完毕， 属于最常见的actor操作，也是效率更高的。 异步意味着系统的各个部分和资源可以得到更有效的利用.
2. !! - 异步消息发送并且同步等待.  bangbang可以表述为fire-and-wait, 消息也是异步处理，但知道消息处理完成之前， !!方法不会返回。 显然这种方式效率不如!的形式， 最好是迫不得已的特定场景下才选用它。
3. !!! - 异步发送消息并返回Future.  bangbangbang与我们executor.submit返回一个future类似， 它发送完消息之后即可返回一个future， 拿到这个future之后，用户自己来决定等待还是先做其它事情。可以描述为fire-and-reply-asynchronously.

为了说明以上说明场景，假设我们有这么一个Actor，它收到数字类型的消息，则直接异步处理； 收到字符串形式的消息， 则将其变为大写，并返回， 该Actor定义如下：

class MyActor extends Actor{
  protected def receive = {
    case n:Int => println("receives number:"+n)
    case msg:String=> self.reply(msg.toUpperCase)
    case _=> println("unexpected message")
  }
}
     

现在，要Actor开工:

val myActor:ActorRef = actorOf[MyActor].start
myActor ! 1                            //receives number:1
println((myActor !! "darren").getOrElse("impossible"))        // DARREN
val future = myActor !!! "darren"
// do something else
future.await
future.result.asInstanceOf[Option[String]] match{
   case Some(str)=> println("converted string:"+str)  // converted string:DARREN
   case None => println("impossible")
}
     

See？ Easy And Fun! 就这么简单， 别在问我其它的了，我也不知道 ：-）

2.1.3. Stop or KillTheActor

Actor干完活， 不能让它在哪儿浪费资源不是？！ 要关闭/停止Actor， 可以采用多种形式， 最常见的就是直接掉actor.stop, 其它几种方式还有:

• Actor接收到特定消息后主动退出. 

  case object ShutdownSignal
  
  class MyActor extends Actor{
    protected def receive = {
   case ShutdownSignal => exit
   case _=> ...
    }
  }
       
• 喂毒药.  每次定义个ShutdownSignal啥的， 还得每次都在Actor实现的时候处理， 真麻烦， 时不？ 其实，可以不这么干， 直接喂你的Actor， 毒死它，哈哈.... 只要myActor ! akka.actor.PoisonPill， 你懂的...
• 通过ActorRegistry统一关闭. Actor.registry.shutdownAll， 你也懂的

2.1.4. Lifecycle Callbacks

除了启动和关闭Actor， Akka Actor还有其它几个有关其生命周期的回掉方法:

• preStart
• postStop
• preRestart
• postRestart

具体在啥时候触发，方法名已经很明了了， 至于用来干什么，也不用多说吧？！做做资源初始化啦， 资源清理啦，这是最常见的用途咯

2.1.5. Other Features

Akka Actor还有些比较有趣的特性，比如运行期间的Actor逻辑互换， 即在运行期间，将当前的处理逻辑替换为新的处理逻辑， 这可以通过Hotswap消息或者become方法来实现， 以Hotswap消息为例(摘自Akka Actor文档):

actor ! HotSwap( self => {
  case message => self.reply("hotswapped body")
})
     

当actor收到这个消息之后，就会使用新的处理逻辑来处理后继的消息。这个新的block会被放到最stack的最上面， 本着“自己的屁股自己擦”的原则，是不是用完后要处理掉那？ become和unbecome可能更能说明这个问题， 有创建就有销毁嘛！

class Swapper extends Actor {
 def receive = {
   case Swap =>
     println("Hi")
     become {
       case Swap =>
         println("Ho")
         unbecome // resets the latest 'become' (just for fun)
     }
 }
}
    

Hi， Ho， Hi，Ho...

2.2. Typed Actor

Typed Actor就是通过强类型的接口约束以及其POJO形式的实现而表现出来的一种Actor形式。Akka框架通过AOP对这些POJO实现相应的接口方法进行拦截和代理， 从而实现跟UntypeActor类似的异步消息处理。 反正方法调用本来也是消息传递的一种嘛。

Typed Actor是根据接口的方法定义的Signature来决定处理语义的， 比如是异步的消息处理，还是异步接收同步返回， 以以下接口为例：

interface RegistrationService {
  void register(User user, Credentials cred)
  User getUserFor(String username)
}
    

返回值为void的方法， 相当于Untyped Actor的！方法， 而有返回值的， 则相当于！！， 至于跟!!!相当的方法定义，就得返回特定的akka.dispatch.Future类型咯。

Typed Actor个人没有啥喜好，所以不做赘述， 感兴趣可以参考 官方文档 。

3. Dispatcher

所有的AkkaActor创建之后，如果没有特殊设置，都会使用Dispatchers$globalExecutorBasedEventDrivenDispatcher$进行运行调度， 从运行日志就可以看出:

14:27:19.659 [main] DEBUG a.d.Dispatchers$globalExecutorBasedEventDrivenDispatcher$ 
   - Starting up Dispatchers$globalExecutorBasedEventDrivenDispatcher$[akka:event-driven:dispatcher:global]
 with throughput [5]

但很多时候， 所有的Actor使用同一个Dispatcher是远远不够的。

要定制相应Actor的运行调度， 可以为其设置特定的Dispatcher实现， 常见的Dispatcher实现在akka.dispatch包下， 如:

• ExecutorBasedEventDrivenDispatcher
• ExecutorBasedEventDrivenWorkStealingDispatcher
• HawtDispatcher

可以通过两种方式为我们的Actor指定自己的Dispatcher， 第一种， 直接在Actor定义中指定：

class MyActor extends Actor{
 self.dispatcher = ...
}
   

使用哪种现有的Dispatcher实现或者自己实现， 可以根据自己的当前环境来决定。 另一种， 可以在Actor的ActorRef上设置:

val myActor = actorOf(..)
myActor.dispatcher = ...
myActor.start
   

只要在Actor启动前设置即可， 个人倾向于这种方式。

4. Fault Tolerance Facility

为了保证系统的健壮性， Akka提供了类似于Erlang的Actor Supervisor的策略， 那个图我就不贴了，就是按照层次， 上层监控下层，如果下层有Actor挂掉了， 那么由上层的Supervisor来决定是重启这个Actor那，还是重启Supervisor当前监控的所有Actor， 所谓的OneForOne和AllForOne策略。

基于Supervisor的策略的基本原则是Let it crash, 当目标Actor crash掉之后， 要么重启它或者相关的actors， 要么正常关闭它。所以， 有几点主要要素要考虑在内， 其一， 需要监控哪些条件来判断目标的Actor有没有crash， 这通常是通过Exception来确定的；其二， 当监控到目标Actor crash之后， 是要重启它，还是直接关闭它？ 其三， 当要重启它的时候，是只重其它那，还是重启跟它相关的所有Actors？ 本着这几个要素，Akka 提供了相应的设施来界定这些关注点。

要实现基于Supervisor的Fault Tolerance管理， Akka中通常采用所谓的声明式的实现方式， 以下是常见的代码：

val supervisor = Supervisor(
  SupervisorConfig(
    AllForOneStrategy(List(classOf[Exception]), 3, 1000),
    Supervise(
      actorOf[MyActor1],
      Permanent) ::
    Supervise(
      actorOf[MyActor2],
      Permanent) ::
    Nil))   
   

Supervisor有一个company object提供了工厂方法来创建和启动一个Supervisor实例，即以上代码种的supervisor， 该工厂方法接受一个SupervisorConfig参数， 而SupervisorConfig参数可以定义Fault Handling的策略，以及要对哪些actor进行监控， 在以上代码中， AllForOneStrategy(List(classOf[Exception]), 3, 1000)定义了当捕获到目标Actor抛出Exception之后， 重启当前supervisor监控的所有actors， 并且最多重启3次， 时间不超出1秒钟。该项配置阐述了我们之前提到的第一和第三要素。SupervisorConfig第二个参数是接受了一组Supervise的List， 每个Supervise定义一个被当前supervisor监控的actor信息， 比如要监控的actor实例引用，以及该actor crash掉之后， 只要一直重启还是只是正常关闭它， 这阐述了我们之前提到的第二个要素。

通过Supervisor创建Supervisor， 最终创建后的实例是已经start的， 不需要显式的start； 不过， 如果需要控制实例创建的时机以及start的时机， 可以使用akka.actor.SupervisorFactory类:

val factory = SupervisorFactory(
   SupervisorConfig(
     AllForOneStrategy(OneForOne, 3, 10, List(classOf[Exception]),
     Supervise(
       myFirstActor,
       Permanent) ::
     Supervise(
       mySecondActor,
       Permanent) ::
     Nil))  
val supervisor = factory.newInstance
supervisor.start // start up all managed server 
   
   

不过，大多数情况下，使用Supervisor就OK啦。

以下是一个完整的实例， 可以一窥声明式的Fault Handling是多么的简单和容易理解：

class MyActor extends Actor {
 // self.setLifeCycle(akka.config.Supervision.permanent)
      override def postRestart(reason: Throwable) {
        println("restart is done.")
      }
      protected def receive = {
        case n: Int => println("receives number:" + n)
        case msg: String => self.reply(msg.toUpperCase)
        case _ => throw new Exception("xxxx")
      }
    }
    import akka.config.Supervision._
    val myActor: ActorRef = actorOf[MyActor].start
    val supervisor = Supervisor(SupervisorConfig(
      AllForOneStrategy(List(classOf[Exception]), 3, 1000),
      Supervise(
        myActor,
        Permanent) ::
        Nil))
 // supervisor.link(myActor)
    myActor ! Array[Byte](0x00)
    myActor ! 111

    TimeUnit.SECONDS.sleep(5)

    supervisor.shutdown
    myActor.stop
   

在myActor ! Array[Byte](0x00)之后， myActor会抛出异常， 但因为有supervisor监控它并重启了它，所以myActor ! 111可以正常处理。 如果不使用声明式的fault handling， 那么需要在supervisor以及被监控actor内部各自声明一些钩子， 以上代码实例中被注视的代码算是一部分， 使用了声明式的方式，这些就都不需要了。

要了解fault handling更多信息，包括编程式的fault handling方式， 可以参考 这里 .

5. Remote Actor

使用就不扯了，简单说一下原理，不过我自己瞎猜的，没看它代码，呵呵，

RemoteActor要实现起来其实很简单， 直接对ActorRef做Proxy， 发送端结点与接收端结点采用不同的逻辑实现， 即接收端结点起相应的TCP服务接收请求， 然后对接收到的请求进行分类并分发给相应的本地Actor， 如果是异步消息，则返回；否则等处理完后， 再将处理结果通过TCP连接发送回发送端结点；发送端结点的Proxy直接发起TCP连接到目标接收端， 之后，只要本地Actor收到请求，就通过该TCP连接将消息发送给目标端TCP Server，server处理逻辑前面已经说了。

在这里， 如果设计好的话， tcp的交互协议可以扩展， tcp的通信序列化协议也可以扩展， 比如， 针对Akka不支持protostuff的现状， 插入自定义的protostuff序列化装置就很好嘛，哈哈

6. Conclusion

AkkaActor相对于其它的实现来说更成熟一些， 但没有最好，只有更好，希望Scala平台的Actor更加成熟，更加快速，更加强大...

开放个人docworks到github

https://github.com/fujohnwang/docworks

docbook maven build archetype project with docbkx maven plugin 

This is my personal docbook project which helps me write a lot of blogs and articles. I had an Ant-based alike project before which helped me complete my first book - "UnveilSpring"(Spring揭秘 In Chinese), but it just works. After I moved to Maven, I found the docbkx maven plugin, so it finally helped me build this archetype project.

To start to use it, you need to know some common locations:

1- src/main/docbook

you put your docbook writings under this directory, sub-directories are allowed. In practice, you should create its own sub-directory for your articles, the reason is with each of your article, it not only contains words but also pictures and other media materials. So usually, before starting to write a new article, you create a sub-folder under src/main/docbook and at least create an "images" sub-folder to hold the images of your article, the real docbook writing will be dropped under this sub-folder directly.

Sounds tedious, don't worry, I create some easy things to help you on this:

mvn -Pcreate

or

mvn -PcreateX -Dmodule=the path you want to create for your article

The former command will create a "tmpdir" under src/main/docbook, and copy article template and necessary images(addons and callout icons) to it, you can rename it after the command run. Of course, if you want to customize the name of the folder to be created, you can run the latter command. OK, that's for src/main/docbook folder.

2- src/main/resources

I have collected necessary resources and put them under this folder, most of the time, you don't need to worry about these things.

3- src/main/assembly

If you wan't to distributed your articles, you can customize assembly descriptor under this folder, of course, using the default is ok.

Since you have know everything necessary, and you may have put some docbook articles under the src/main/docbook folder, run "mvn package", after that, you can get your generated document under "target/docs/html" folder, What? Build Error? Find out the reason yourself by focusing on your docbook document structure and elements ;-)

GL & HF

Prototyping An Actor In Java

Prototyping An Actor In Java

王福强(Darren.Wang)

<fujohnwang@gmail.com>

---

Table of Contents

1. Actor with Same Instance Reference of Action(Failed)

2. Actor With Prototype Scope Action(Workable)

3. Actor With State Copy Between Prototype Scope Actions

4. Raw Thread Based Actor Prototyping (Workable)

4.1. Another Version

5. Conclusion

I have always try to implement an actor like the ones of Erlang or Scala in java, and had tried several times with different strategies. The former attempts seems naive, and I will not demonstrate it here, but the ideas from those naive prototypes still hold.

Note

• I hope you know something about the actor model before continuing to read this.
• We mainly focus on "fire-and-get" pattern, the "fire-and-forget" pattern is much easier to implement.
• To simply the prototyping process, I will left any generic type design behind, hope u can polish it so that it can be your own tools.

Before start, we need to make something clear. In Java, the concurrency is modeled in Task-Based Concurrency, We run different logic which is modeled as tasks to run in parallel. But The actor is different, it belongs to the category of Data-Based Concurrency, to bridge the differences between this two concurrency models, I will declare an interface to confine the process logic into it:

public interface ActorClosure<T> extends Runnable {
     void sinkEvent(T event);
 }

  

ActorClosure will accept different events each time but run them one by one with same process logic. That's why a sinkeEvent() method is declared.

Besides, we also give out an stub for our demonstration:

public class ActorClosureStub implements ActorClosure<String> {
    private String event;
    
    public void run() {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
         // log the exception
            return;
        }
        System.out.println("received event:" + event);
    }
    public void sinkEvent(String event) {
        this.event = event;
    }
}
  

OK, background is done, let's start with our java actor prototyping play...

1. Actor with Same Instance Reference of Action(Failed)

Since each event that send to the actor should be processed by one processing logic, At first, we will just let our actor to use only one ActorClosure to process all of the events that's sent to it. So below is the code :

public class FailedActorWithSameReference {

    private ExecutorService      scheduler = Executors.newFixedThreadPool(1);
    private ActorClosure<String> action    = new ActorClosureStub();

    public Future<?> bang(String event) throws InterruptedException {
        action.sinkEvent(event);
        return scheduler.submit(action, null);
    }

    public void terminate() {
        scheduler.shutdown();
    }

    public static void main(String[] args) throws Exception {
        FailedActorWithSameReference actor = new FailedActorWithSameReference();
        Future<?> f1 = actor.bang("a");
        Future<?> f2 = actor.bang("b");
        f1.get();
        f2.get();
        actor.terminate();
    }
}
   

In fact, no need to run this piece of code, we can see that it won't work properly, why? Because we didn't handle the concurrency operation on the ActorClosure properly, the latter bang(!) will replace or corrupt the state in the ActorClosure, If you run the above piece of code, it will print "b" twice, but we do expect "a" and "b" in sequence. But the problem really is the bad concurrency control on ActorClosure? No, If we add concurrency control on ActorClosure, then we go wrong way in the process of modeling an actor.

So, no concurrency control on ActorClosure, how to avoid the state corruption?

2. Actor With Prototype Scope Action(Workable)

We of course can allocate a new ActorClosure instance for each event, in this way, the state of ActorClosures and their events will be confined into their own boundary without leak and interference.

With this in mind, we got code piece below:

/**
 * This actor implementation just simply works, but it can't fully simulate the
 * exact behavior of an actor, e.g. since we will create new action closure for
 * each event, the states of the action closure can't be kept in the time-line
 * of the actor, maybe we should introduce some copy mechanism to complement
 * this.
 * 
 * @author fujohnwang
 * @since 1.0
 */
public class ActorWithPrototypeScopeActionClosure {

    private ExecutorService                       scheduler  = Executors.newFixedThreadPool(1);

    private Class<? extends ActorClosure<String>> actionType = ActorClosureStub.class;

    public Future<?> bang(String e) throws Exception {
        ActorClosure<String> action = actionType.newInstance();
        action.sinkEvent(e);
        return scheduler.submit(action, null);
    }

    public void setActionType(Class<? extends ActorClosure<String>> actionType) {
        this.actionType = actionType;
    }

    public Class<? extends ActorClosure<String>> getActionType() {
        return actionType;
    }

    public void terminate() {
        scheduler.shutdown();
    }

    public static void main(String[] args) throws Exception {
        ActorWithPrototypeScopeActionClosure actor = new ActorWithPrototypeScopeActionClosure();
        Future<?> f1 = actor.bang("a");
        Future<?> f2 = actor.bang("b");
        Future<?> f3 = actor.bang("c");

        f3.get();
        f2.get();
        f1.get();

        actor.terminate();
    }
}
   

If we run the above piece of code, the result will be as we expected. Seems it work, but wait, it's not perfect. Why? (Why again ha?)

Since we wrap each event(the data) into a ActorClosureStub instance(the task), we lost the smooth timeline state of the processsing. Simply put, the ActorClosureStub runs first has no correlation with other ActorClosureStubs. But in actor's semantics, they should be one. So although this prototyping seems work, but it still have improvement space.

3. Actor With State Copy Between Prototype Scope Actions

So I hope to copy the state between each ActorClosure after they run, in this way, the 1st ActorClosure will pass its state to the 2nd ActorClosure before the 2nd ActorClosure will run, and the 2nd ActorClosure will pass its state to the 3rd ActorClosure before the 3rd ActorClosure will run, and so on.

To make this happen, we can recall the hooks the ThreadPoolExecutor has, e.g. afterExecute(), beforeExecute(), furthermore, we can even directly override the execute() of it. Anyway, we can write such-alike code:

/**
 * Although we seek to extend {@link ActorWithPrototypeScopeActionClosure} to
 * enable state copy between the action closures, BUT it seems that the Executor
 * implementation doesn't allow us to go this way.<br>
 * This Actor DOESN'T Work!!!!
 * 
 * @author fujohnwang
 * @since 1.0
 */
public class ActorWithStateCopy {

    private ActorClosure<String>                           lastClosure;
    private ActionClosureStateCopier<ActorClosure<String>> stateCopier = new ActionClosureStateCopier<ActorClosure<String>>() {
                                                                           public void copy(ActorClosure<String> from,
                                                                                            ActorClosure<String> to) {
                                                                               System.out
                                                                                       .println("do copy if necessary");
                                                                           }

                                                                       };

    private ExecutorService                                scheduler   = new ThreadPoolExecutor(
                                                                               1,
                                                                               1,
                                                                               60,
                                                                               TimeUnit.SECONDS,
                                                                               new ArrayBlockingQueue<Runnable>(
                                                                                       10)) {

//                                                                           @Override
//                                                                           protected void afterExecute(Runnable arg0,
//                                                                                                       Throwable arg1) {
//                                                                               super.afterExecute(
//                                                                                       arg0, arg1);
//                                                                           }
//
//                                                                           @Override
//                                                                           protected void beforeExecute(Thread arg0,
//                                                                                                        Runnable arg1) {
//                                                                               super.beforeExecute(
//                                                                                       arg0, arg1);
//                                                                           }

                                                                           @SuppressWarnings("unchecked")
                                                                           @Override
                                                                           public void execute(Runnable command) {
                                                                               ActorClosure<String> r = (ActorClosure<String>) command;
                                                                               if (lastClosure != null
                                                                                       && stateCopier != null) {
                                                                                   System.out
                                                                                           .println("copy state before executing");
                                                                                   stateCopier
                                                                                           .copy(lastClosure,
                                                                                                   r);
                                                                               }
                                                                               super.execute(r);

                                                                               lastClosure = r;
                                                                           }

                                                                       };

    private Class<? extends ActorClosure<String>>          actionType  = ActorClosureStub.class;

    public Future<?> bang(String e) throws Exception {
        ActorClosure<String> action = actionType.newInstance();
        action.sinkEvent(e);
        return scheduler.submit(action, null);
    }

    public void setActionType(Class<? extends ActorClosure<String>> actionType) {
        this.actionType = actionType;
    }

    public Class<? extends ActorClosure<String>> getActionType() {
        return actionType;
    }

    public void terminate() {
        scheduler.shutdown();
    }

    /**
     * @param args
     */
    public static void main(String[] args) throws Exception {
        ActorWithStateCopy actor = new ActorWithStateCopy();
        Future<?> f1 = actor.bang("a");
        Future<?> f2 = actor.bang("b");
        Future<?> f3 = actor.bang("c");

        f3.get();
        f2.get();
        f1.get();

        actor.terminate();
    }

}
   

Let's run the code. Oops, something goes wrong:

Exception in thread "main" java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to cn.spring21.sandbox.actor.ActorClosure
 at cn.spring21.sandbox.actor.ActorWithStateCopy$2.execute(ActorWithStateCopy.java:55)
 at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:85)
 at cn.spring21.sandbox.actor.ActorWithStateCopy.bang(ActorWithStateCopy.java:76)
 at cn.spring21.sandbox.actor.ActorWithStateCopy.main(ActorWithStateCopy.java:96) 
   
   

See the problem? Wow, ThreadPoolExecutor really do some dirty things there, and I tries to bypass it but finally, I found I couldn't :-(

4. Raw Thread Based Actor Prototyping (Workable)

I fall back to a raw thread solution for the actor. The final code is listed here:

public class ImprovedThreadBasedActor<T> extends Thread {

    private static final Logger            logger  = LoggerFactory
                                                           .getLogger(ImprovedThreadBasedActor.class);

    private BlockingQueue<EventWrapper<T>> mailbox;
    private volatile boolean               running = true;
    private ActorClosure<T>                action;

    public ImprovedThreadBasedActor(ActorClosure<T> action) {
        this(action, 100);
    }

    public ImprovedThreadBasedActor(ActorClosure<T> action, int mailboxSize) {
        Validate.notNull(action);

        mailbox = new ArrayBlockingQueue<EventWrapper<T>>(mailboxSize <= 0 ? 100 : mailboxSize);
        this.action = action;

    }

    public Future<?> sendAndAsyncWait(T event) throws Exception {
        if (!running) {
            throw new Exception("the actor is down");
        }
        FutureTask<Object> future = new FutureTask<Object>(new Runnable() {
            public void run() {
            }
        }, true);
        EventWrapper<T> wrapper = new EventWrapper<T>(event, future);
        mailbox.put(wrapper);
        return future;
    }

    @Override
    public void run() {
        while (running) {
            try {
                EventWrapper<T> wrapper = mailbox.take();
                try {
                    action.sinkEvent(wrapper.getEvent());
                    action.run();
                    wrapper.getFuture().run();
                } catch (Exception e) {
                    logger.error("exception in actor execution which will stop the actor:\n{}",
                            ExceptionUtils.getFullStackTrace(e));
                    break;
                }
            } catch (InterruptedException e) {
                logger.info("ImprovedThreadBasedActor running thread is interrupted:{}",
                        ExceptionUtils.getFullStackTrace(e));
                continue;
            }
        }
        logger.info("ImprovedThreadBasedActor is down.");
    }

    public void terminate() {
        running = false;
        interrupt();
    }

    public static void main(String[] args) throws Exception {
        ImprovedThreadBasedActor<String> actor = new ImprovedThreadBasedActor<String>(
                new ActorClosureStub(), 10);
        actor.start();

        Future<?> f1 = actor.sendAndAsyncWait("a");
        Future<?> f2 = actor.sendAndAsyncWait("b");
        Future<?> f3 = actor.sendAndAsyncWait("c");
        f3.get();
        System.out.println("f2 is done? " + f2.isDone());
        f2.get();
        System.out.println("f1 is done? " + f2.isDone());
        f1.get();
        System.out.println("stop the actor.");
        actor.terminate();
    }
}

   

No explanation on this, you can explore it yourself. There are several tricky things in the code, I hope you can figure out the reason. GL and HF.

4.1. Another Version

/**
 * This is a heavy weight thread based actor implementation.<br>
 * 
 * @author fujohnwang
 * @param <T>, the event type to be processed
 * @param <R>, result type after processing the event.
 */
public class ThreadActor<T, R> extends Thread {
    private transient final Logger            logger  = LoggerFactory.getLogger(this.getClass());

    private volatile boolean                  running = true;
    private BlockingQueue<EventWrapper<T, R>> mailbox;
    private Reaction<T, R>                    action;

    public ThreadActor(Reaction<T, R> action, int mailboxSize) {
        this.action = action;
        this.mailbox = new ArrayBlockingQueue<EventWrapper<T, R>>((mailboxSize <= 0) ? 100
                : mailboxSize);
    }

    public Future<R> sendAndAsycAwait(T event) throws Exception {
        if (!running) {
            throw new Exception("the actor is down");
        }
        FutureProxy<R> f = new FutureProxy<R>(new FutureTask<R>(new Callable<R>() {
            public R call() throws Exception {
                return null;
            }
        }));
        mailbox.put(new EventWrapper<T, R>(event, f));
        return f;
    }

    @Override
    public void run() {
        while (running) {
            try {
                EventWrapper<T, R> e = mailbox.take();
                action.sink(e.getEvent());
                try {
                    R r = action.call();
                    e.getFuture().setResult(r);
                } catch (Exception e1) {
                    e.getFuture().setCause(e1);
                } finally {
                    e.getFuture().getDelegate().run();
                }
            } catch (InterruptedException e) {
                continue;
            }
        }
        logger.info("actor shutdown");
    }

    public void terminate() {
        running = false;
        interrupt();
    }

    public static void main(String args[]) {
        ThreadActor<String, Boolean> actor = new ThreadActor<String, Boolean>(new ReactionStub(),
                10);
        actor.start();

        try {
            for (int i = 0; i < 10; i++) {
                try {
                    System.out.println(actor.sendAndAsycAwait(String.valueOf(i)).get());
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        } finally {
            actor.terminate();
        }
    }
}

class ReactionStub implements Reaction<String, Boolean> {

    private long   counter;
    private String event;

    public Boolean call() throws Exception {
        try {
            if (counter % 2 == 0) {
                System.out.println("message received:" + event);
                return true;
            } else {
                throw new Exception("sample exception");
            }
        } finally {
            counter += 1;
        }

    }

    public void sink(String event) {
        this.event = event;
    }

}
    

优劣不解释。

5. Conclusion

The above words just a play-around, don't treat it too serious. It's not rocket science, If you are looking for such thing, get away. These words are not for you.

Caution: The last method still has gotchas, Improvement is still in progress