如何設(shè)計(jì)一個(gè)實(shí)時(shí)流計(jì)算系統(tǒng)？
實(shí)時(shí)流計(jì)算的場景歸納起來多半是：業(yè)務(wù)系統(tǒng)根據(jù)實(shí)時(shí)的操作，不斷生成事件（消息/調(diào)用），然后引起一系列的處理分析，這個(gè)過程是分散在多臺(tái)計(jì)算機(jī)上并行完成的，看上去就像事件連續(xù)不斷的流經(jīng)多個(gè)計(jì)算節(jié)點(diǎn)處理，形成一個(gè)實(shí)時(shí)流計(jì)算系統(tǒng)。
市場上流計(jì)算產(chǎn)品有很多，主要是通過消息中樞結(jié)合工人模式實(shí)現(xiàn)，大致過程如下：
1、開發(fā)者實(shí)現(xiàn)好流程輸入輸出節(jié)點(diǎn)邏輯，上傳job到任務(wù)生產(chǎn)者
2、任務(wù)生產(chǎn)者將任務(wù)發(fā)送到zookeeper，然后監(jiān)控任務(wù)狀態(tài)
3、任務(wù)消費(fèi)者從zookeeper上獲取任務(wù)
4、任務(wù)消費(fèi)者啟動(dòng)多個(gè)工人進(jìn)程，每個(gè)進(jìn)程又啟動(dòng)多個(gè)線程執(zhí)行任務(wù)
5、工人之間通過zeroMQ交互
我們看看如何做一個(gè)簡單的流計(jì)算系統(tǒng)，做法跟上面有些不同：
1、首先不過多依賴zookeerper,任務(wù)的分配最好直接給到工人，并能直接監(jiān)控工人完成狀態(tài)，這樣效率會(huì)更高。
2、工人之間直接通訊，不依賴zeroMQ轉(zhuǎn)發(fā)。
3、并行管理扁平化，多進(jìn)程下再分多線程意義不大，增加管理成本，實(shí)際上一臺(tái)機(jī)器8個(gè)進(jìn)程，每個(gè)進(jìn)程再開8個(gè)線程，總體跟8-10個(gè)進(jìn)程或者線程的效果差不多（數(shù)量視機(jī)器性能不同）。
4、做成一個(gè)流計(jì)算系統(tǒng)，而不是平臺(tái)。
這里我們借助fourinone提供的api和框架去實(shí)現(xiàn)，第一次使用可以參考分布式計(jì)算上手demo指南，開發(fā)包下載地址 http://code.google.com/p/fourinone/
大致思路：用工頭去做任務(wù)生產(chǎn)和分配，用工人去做任務(wù)執(zhí)行，為了達(dá)到流的效果，需要在工人里面調(diào)用工頭的方式，將多個(gè)工人節(jié)點(diǎn)串起來。
下面程序演示了連續(xù)多個(gè)消息先發(fā)到一個(gè)工人節(jié)點(diǎn)A處理，然后再發(fā)到兩個(gè)工人節(jié)點(diǎn)B并行處理的流計(jì)算過程，并且獲取到最后處理結(jié)果打印輸出（如果不需要獲取結(jié)果可以直接返回）。
- StreamCtorA：工頭A實(shí)現(xiàn)，它獲取到線上工人A，然后將消息發(fā)給它處理，并輪循等待結(jié)果。工頭A的main函數(shù)模擬了多個(gè)消息的連續(xù)調(diào)用。
- StreamWorkerA：工人A實(shí)現(xiàn)，它接收到工頭A的消息進(jìn)行處理，然后創(chuàng)建一個(gè)工頭B，通過工頭B將結(jié)果同時(shí)發(fā)給兩個(gè)工人B處理，然后將結(jié)果返回工頭A。
- StreamCtorB：工頭B實(shí)現(xiàn)，它獲取到線上兩個(gè)工人B，調(diào)用doTaskBatch等待兩個(gè)工人處理完成，然后返回結(jié)果給工人A。
- StreamWorkerB：工人B實(shí)現(xiàn)，它接收到任務(wù)消息后模擬處理后返回結(jié)果。
運(yùn)行步驟（在本地模擬）：
1、啟動(dòng)ParkServerDemo（它的IP端口已經(jīng)在配置文件指定）
- java -cp fourinone.jar; ParkServerDemo
2、啟動(dòng)工人A
- java -cp fourinone.jar; StreamWorkerA localhost 2008
3、啟動(dòng)兩個(gè)工人B
- java -cp fourinone.jar; StreamWorkerB localhost 2009
- java -cp fourinone.jar; StreamWorkerB localhost 2010
4、啟動(dòng)工頭A
- java -cp fourinone.jar; StreamCtorA
多機(jī)部署說明：StreamCtorA可以單獨(dú)部署一臺(tái)機(jī)器，StreamWorkerA和StreamCtorB部署一臺(tái)機(jī)器，兩個(gè)StreamWorkerB可以部署兩臺(tái)機(jī)器。
總結(jié)：計(jì)算平臺(tái)和計(jì)算系統(tǒng)的區(qū)別
如果我們只有幾臺(tái)機(jī)器，但是每天有人開發(fā)不同的流處理應(yīng)用要在這幾臺(tái)機(jī)器上運(yùn)行，我們需要一個(gè)計(jì)算平臺(tái)來管理好job，讓開發(fā)者按照規(guī)范配置好流程和運(yùn)行時(shí)節(jié)點(diǎn)申請，打包成job上傳，然后平臺(tái)根據(jù)每個(gè)job配置動(dòng)態(tài)分配資源依次執(zhí)行每個(gè)job內(nèi)容。
如果我們的幾臺(tái)機(jī)器只為一個(gè)流處理業(yè)務(wù)服務(wù)，比如實(shí)時(shí)營銷，我們需要一個(gè)流計(jì)算系統(tǒng)，按照業(yè)務(wù)流程部署好計(jì)算節(jié)點(diǎn)即可，不需要運(yùn)行多個(gè)job和動(dòng)態(tài)分配資源，按照計(jì)算平臺(tái)的方式做只會(huì)增加復(fù)雜性，開發(fā)者也不清楚每臺(tái)機(jī)器上到底運(yùn)行了什么邏輯。
如果你想實(shí)現(xiàn)一個(gè)計(jì)算平臺(tái)，可以參考動(dòng)態(tài)部署和進(jìn)程管理功能（開發(fā)包內(nèi)有指南）
//完整源碼
// ParkServerDemo
- 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
- view sourceprint?
- 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");
- }
- }
