實時流計算的場景歸納起來多半是： 
業務系統根據實時的操作，不斷生成事件（訊息/呼叫），然後引起一系列的處理分析，這個過程是分散在多臺計算機上並行完成的，看上去就像事件連續不斷的流經多個計算節點處理，形成一個實時流計算系統。
市場上流計算產品有很多，主要是通過訊息中樞結合工人模式實現，大致過程如下： 
1、開發者實現好流程輸入輸出節點邏輯，上傳job到任務生產者 
2、任務生產者將任務傳送到zookeeper，然後監控任務狀態 
3、任務消費者從zookeeper上獲取任務 
4、任務消費者啟動多個工人程序，每個程序又啟動多個執行緒執行任務 
5、工人之間通過zeroMQ互動 

我們看看如何做一個簡單的流計算系統，做法跟上面有些不同： 

import com.fourinone.BeanContext;
public class ParkServerDemo
{
public static void main(String[] args)
{
BeanContext.startPark();
}
}

//StreamCtorA 

import com.fourinone.Contractor;
import com.fourinone.WareHouse;
import com.fourinone.WorkerLocal;
import java.util.ArrayList;

public class StreamCtorA extends Contractor
{
 public WareHouse giveTask(WareHouse inhouse)
 {
  WorkerLocal[] wks = getWaitingWorkers("StreamWorkerA");
  System.out.println("wks.length:"+wks.length);
  
  WareHouse result = wks[0].doTask(inhouse);
  while(true){
   if(result.getStatus()!=WareHouse.NOTREADY)
   {
    break;
   }
  }
  return result;
 }
 
 public static void main(String[] args)
 {
  StreamCtorA sc = new StreamCtorA();
  for(int i=0;i<10;i++){
    WareHouse msg = new WareHouse();
    msg.put("msg","hello"+i);
    WareHouse wh = sc.giveTask(msg);
    System.out.println(wh);
  }
  sc.exit();
 }
}

//StreamWorkerA 

import com.fourinone.MigrantWorker;
import com.fourinone.WareHouse;

public class StreamWorkerA extends MigrantWorker
{
 public WareHouse doTask(WareHouse inhouse)
 {
  System.out.println(inhouse);
  //do something
  StreamCtorB sc = new StreamCtorB();
  WareHouse msg = new WareHouse();
  msg.put("msg",inhouse.getString("msg")+",from StreamWorkerA");
  WareHouse wh = sc.giveTask(msg);
  sc.exit();
  
  return wh;
 }
 
 public static void main(String[] args)
 {
  StreamWorkerA wd = new StreamWorkerA();
  wd.waitWorking(args[0],Integer.parseInt(args[1]),"StreamWorkerA");
 }
}

//StreamCtorB 

import com.fourinone.Contractor;
import com.fourinone.WareHouse;
import com.fourinone.WorkerLocal;
import java.util.ArrayList;

public class StreamCtorB extends Contractor
{
 public WareHouse giveTask(WareHouse inhouse)
 {
  WorkerLocal[] wks = getWaitingWorkers("StreamWorkerB");
  System.out.println("wks.length:"+wks.length);
  
  WareHouse[] hmarr = doTaskBatch(wks, inhouse);
  
  WareHouse result = new WareHouse();
  result.put("B1",hmarr[0]);
  result.put("B2",hmarr[1]);
  
  return result;
 }
}

//StreamWorkerB 

import com.fourinone.MigrantWorker;
import com.fourinone.WareHouse;

public class StreamWorkerB extends MigrantWorker
{
 public WareHouse doTask(WareHouse inhouse)
 {
  System.out.println(inhouse);
  //do something
  inhouse.put("msg",inhouse.getString("msg")+",from StreamWorkerB");
  return inhouse;
 }
 
 public static void main(String[] args)
 {
  StreamWorkerB wd = new StreamWorkerB();
  wd.waitWorking(args[0],Integer.parseInt(args[1]),"StreamWorkerB");
 }
}