实时流计算的场景归纳起来多半是:
业务系统根据实时的操作,不断生成事件(消息/调用),然后引起一系列的处理分析,这个过程是分散在多台计算机上并行完成的,看上去就像事件连续不断的流经多个计算节点处理,形成一个实时流计算系统。
市场上流计算产品有很多,主要是通过消息中枢结合工人模式实现,大致过程如下:
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");
}
}
分享到:
相关推荐
通向“亿万级”实时流计算之路,应用领域:时延敏感的应用领域
基于事件的流计算系统的设计和实现,周超,邹华,随着互联网的不断发展,实时数据逐渐体现出巨大价值。流计算技术被应用到实时数据的处理中,源源不断的数据作为输入,实时处理,
计算平台事业部 付空在2018云栖大会·上海峰会中做了题为《基于流计算构建实时大数据处理系统》的分享,就为什么要用流计算、为什么要用阿里云流计算、如何用流计算等方面的内容做了深入的分析。
Streaming data is a big deal in big data these days. As more and more businesses seek to tame the massive unbounded data sets that pervade our world, streaming systems have finally reached a level of ...
#资源达人分享计划#
【资源说明】 1、该资源包括项目的全部源码,下载可以直接使用! 2、本项目适合作为计算机、数学、电子信息等专业的课程设计、期末大作业和毕设项目,...基于Kafka+Storm+HBase风电行业实时流计算系统源码+项目说明.zip
计算机系统结构论文--计算机系统结构课程设计-数据流计算机 http://janelinbj.blog.163.com/
基于Spark流计算框架的银行实时存贷款规模系统设计与实现.pdf
运用实时计算技术、Web 技术构建一个图书商城的促销活动大数据看板,实现以下功能: (1)界面每 3 秒刷新一次,如果能做到 2 秒或 1 秒刷新一次,可加分; (2)展示当前订单的已处理速度,单位为“条/秒”; (3)...
本文将按照如下四个阶段分享微博实时流计算平台的搭建历程,以及在创建过程中的一些问题和解决方案:实时流计算平台初建实时流计算平台发展总结DQRA设计模式初入实时流计算首先介绍一下我们实时流计算平台开发历程:...
本次演讲我们以滴滴大数据实时监控系统为例,介绍滴滴实时计算系统架构所面临的挑战及相应解决方案。 滴滴大数据实时监控系统服务于滴滴各个业务线,对关键业务指标的变化进行实时监控。研发、产品、运营人员可以...
密码,到期日期) 四、 体系结构设计 设计阶段: 顶层数据流图: 储户 银行储蓄系统 储户 存款单 取款单 存款单 利息清单 银行模拟储蓄系统 用户管理 帐户管理 利息管理 修 改 用 户 添 加 用 户 删 除 用 户 用 户 ...
基于flink框架写的大数据系统, 框架:flink 适宜人群:可以作为计算机专业的课程设计或者是毕业设计
这一资源不仅为本科课程设计、毕业设计提供了丰富的实践内容,同时也为深度学习算法学习者提供了一个深入了解车牌识别技术的平台。 此系统通过捕获实时视频流,运用OpenCV的图像处理和计算机视觉技术,对视频中的...
【资源说明】 1、该资源内项目代码都是经过测试运行成功,功能正常的情况下才上传的,请...3、不仅适合小白学习实战练习,也可作为大作业、课程设计、毕设项目、初期项目立项演示等,欢迎下载,互相学习,共同进步!
1、课程中完整开发3个Storm项目,均为企业实际项目,其中一个是完全由Storm Trident开发。 项目源码均可以直接运行,也可直接用于商用或企业。 2、Storm全面、系统、深入讲解 3、注重实践,对较抽象难懂的技术点如...
计算机毕业设计_asp.net基于工作流引擎的系统框架设计开发(源代码+)__毕设源码实例.zip计算机毕业设计_asp.net基于工作流引擎的系统框架设计开发(源代码+)__毕设源码实例.zip计算机毕业设计_asp.net基于工作流引擎的...
强流微束斑电子束系统的研究和设计计算,郭小立,唐天同,本文研究了强流微束斑电子束的形成问题。提出了一种采用新型磁透镜的电磁复合聚焦系统,这种新型电子束系统的轴上磁场在电子束束
《基于工作流引擎的系统框架设计开发——工作流引擎子系统》是采用Visual C#以及Microsoft SQL Server2000来开发的。系统主要由四个模块组成,他们分别是:状态图管理模块、任务列表模块、任务指派模块、任务提交...