`

如何设计一个实时流计算系统

阅读更多
实时流计算的场景归纳起来多半是:
业务系统根据实时的操作,不断生成事件(消息/调用),然后引起一系列的处理分析,这个过程是分散在多台计算机上并行完成的,看上去就像事件连续不断的流经多个计算节点处理,形成一个实时流计算系统。

市场上流计算产品有很多,主要是通过消息中枢结合工人模式实现,大致过程如下:
1、开发者实现好流程输入输出节点逻辑,上传job到任务生产者
2、任务生产者将任务发送到zookeeper,然后监控任务状态
3、任务消费者从zookeeper上获取任务
4、任务消费者启动多个工人进程,每个进程又启动多个线程执行任务
5、工人之间通过zeroMQ交互

我们看看如何做一个简单的流计算系统,做法跟上面有些不同:
1、首先不过多依赖zookeerper,任务的分配最好直接给到工人,并能直接监控工人完成状态,这样效率会更高。
2、工人之间直接通讯,不依赖zeroMQ转发。
3、并行管理扁平化,多进程下再分多线程意义不大,增加管理成本,实际上一台机器8个进程,每个进程再开8个线程,总体跟8-10个进程或者线程的效果差不多(数量视机器性能不同)。
4、做成一个流计算系统,而不是平台。

这里我们借助fourinone提供的api和框架去实现,第一次使用可以参考分布式计算上手demo指南,开发包下载地址 http://code.google.com/p/fourinone/

大致思路:用工头去做任务生产和分配,用工人去做任务执行,为了达到流的效果,需要在工人里面调用工头的方式,将多个工人节点串起来,形成一个计算拓扑图。

下面程序演示了连续多个消息先发到一个工人节点A处理,然后再发到两个工人节点B并行处理的流计算过程,并且获取到最后处理结果打印输出(如果不需要获取结果可以直接返回)。

StreamCtorA:工头A实现,它获取到线上工人A,然后将消息发给它处理,并轮循等待结果。工头A的main函数模拟了多个消息的连续调用。

StreamWorkerA:工人A实现,它接收到工头A的消息进行处理,然后创建一个工头B,通过工头B将结果同时发给两个工人B处理,然后将结果返回工头A。

StreamCtorB:工头B实现,它获取到线上两个工人B,调用doTaskBatch等待两个工人处理完成,然后返回结果给工人A。

StreamWorkerB:工人B实现,它接收到任务消息后模拟处理后返回结果。

运行步骤(在本地模拟):
1、启动ParkServerDemo(它的IP端口已经在配置文件指定)
java -cp fourinone.jar; ParkServerDemo

2、启动工人A
java  -cp fourinone.jar; StreamWorkerA localhost 2008

3、启动两个工人B
java  -cp fourinone.jar; StreamWorkerB localhost 2009
java  -cp fourinone.jar; StreamWorkerB localhost 2010

4、启动工头A
java  -cp fourinone.jar; StreamCtorA

多机部署说明:StreamCtorA可以单独部署一台机器,StreamWorkerA和StreamCtorB部署一台机器,两个StreamWorkerB可以部署两台机器。

总结:计算平台和计算系统的区别
如果我们只有几台机器,但是每天有人开发不同的流处理应用要在这几台机器上运行,我们需要一个计算平台来管理好job,让开发者按照规范配置好流程和运行时节点申请,打包成job上传,然后平台根据每个job配置动态分配资源依次执行每个job内容。
如果我们的几台机器只为一个流处理业务服务,比如实时营销,我们需要一个流计算系统,按照业务流程部署好计算节点即可,不需要运行多个job和动态分配资源,按照计算平台的方式做只会增加复杂性,开发者也不清楚每台机器上到底运行了什么逻辑。
如果你想实现一个计算平台,可以参考动态部署和进程管理功能(开发包内有指南)


//完整源码
// 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
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");
 }
} 
0
3
分享到:
评论
6 楼 swingfish 2014-03-24  
fourinone 写道
swingfish 写道
这个代码放在jdk1.6上编译出错,StreamWorkerA.doTask()里的sc.exit()编译错误:The method exit() from the type ContractorParallel is not visible

3.0版后都是jdk1.7环境下,旧版本没有exit方法


谢谢!
5 楼 fourinone 2014-03-24  
swingfish 写道
这个代码放在jdk1.6上编译出错,StreamWorkerA.doTask()里的sc.exit()编译错误:The method exit() from the type ContractorParallel is not visible

3.0版后都是jdk1.7环境下,旧版本没有exit方法
4 楼 swingfish 2014-03-24  
这个代码放在jdk1.6上编译出错,StreamWorkerA.doTask()里的sc.exit()编译错误:The method exit() from the type ContractorParallel is not visible
3 楼 zhangzhenjj 2013-08-14  
storm?
2 楼 fourinone 2013-08-12  
skzr.org 写道
我想知道你这个东西的性能怎样?

技术博客里有测试报告,可以找找
1 楼 skzr.org 2013-08-12  
我想知道你这个东西的性能怎样?

相关推荐

Global site tag (gtag.js) - Google Analytics