JSTORM使用笔记

httpwwwcnblogscomantispamp4182210html使用笔记安装部署简单快速的传输层框架安装如下应该是的包吧安装步骤如下针对大型分布式系统提供配置维护名字服务分布式同步组服务等可以保证顺序性客户端的更新请求都会被顺序处理原子性更新操作要不成功要不失败一致性客户端不论连接到那个服务端展现给它的都是同一个视图可靠性更新会被持久化实时性对于每个客户端他的系统视图都是最新的在中有几种角色Leader发起投票和决议更新系统状态Follower响应客户端请求参与投票Observer不参与投票只同步Leader状态Client发起请求在启动之前需要在下编写配置文件里面的内容包括tickTime心跳间隔initLimitFollower和Leader之间建立连接的最大心跳数syncLimitFollower和Leader之间通信时限dataDir数据目录dataLogDir日志目录minSessionTimeout最小会话时间默认tickTime2maxSessionTimeout最大会话时间默认tickTime20maxClientCnxns客户端数量clientPort监听客户端连接的端口serverNYYYYAB其中N为服务器编号YYYY是服务器的IP地址A是Leader和Follower通信端口B为选举端口在单机的时候可以直接将修改为然后使用启动服务即可如果报错没有目录手动创建即可现在用或者是就能看到在监听指定的端口那么现在起来了参考httpblogcsdnnetshenlan211314articledetails6170717httpblogcsdnnetshenlan211314articledetails6170717httpblogcsdnnethikevinarticledetails7089358httpblogcsdnnethikevinarticledetails7089358httpapachedatagurucnzookeeperzookeeper346下载地址httpapachedatagurucnzookeeperzookeeper346该系统是阿里巴巴在对做了重写和优化在里面能运行的在里面也能运行该系统擅长执行实时计算而且基本上都在内存中搞定进入正题中有如下几种角色spout源头bolt处理器topology由处理器源头组成的拓扑网络每条边就是一个订阅关系tuple数据worker执行进程task执行线程nimbus分发代码任务监控集群运行状态supervisor监听nimbus的指令接收分发代码和任务并执行是用来管理的下面来看中的常用配置stormzookeeperserverszookeeper集群地址stormzookeeperrootzookeeper中storm的根目录位置stormlocaldir用来存放配置文件JAR等stormmessagingnettytransferasyncbatch在使用Netty的时候设置是否一个batch中会有多个消息javalibrarypath本地库的加载地址比如zeromqjzmq等supervisorslotsportssupervisor节点上的worker使用的端口号列表supervisorenablecgroup是否使用cgroups来做资源隔离topologybuffersizelimited是否限制内存如果不限制将使用LinkedBlockingDequetopologyperformancemetrics是否开启监控topologyalimonitormetricspost是否将监控数据发送给AliMonitortopologyenableclassloader默认禁用了用户自定义的类加载器workermemorysizeworker的内存大小在把配置搞正确之后就可以用中的脚本来启动节点服务了参考httpsgithubcomalibabajstormwikiE5A682E4BD95E5AE89E8A385httpsgithubcomalibabajstormwikiE5A682E4BD95EhttpsgithubcomalibabajstormwikiE5A682E4BD95E5AE89E8A3855AE89E8A385httpifevecomgettingstartedwithstorm5storm编程入门httpifevecomgettingstartedwithstorm5的架构结构和的很像整体看来如下负责控制提交任务负责执行任务为了做实时计算你需要建立由计算节点组成的图在上的httpxumingmingsinaappcom647twitterstormcodeanalysistopologyexecution的生命周期如下上传代码并做校验nimbusinbox建立本地目录stormdisttopologyid建立zookeeper上的心跳目录计算topology的工作量parallelismhint分配taskid并写入zookeeper把task分配给supervisor执行在supervisor中定时检查是否有新的task下载新代码删除老代码剩下的工作交个小弟worker在worker中把task拿到看里面有哪些spoutBolt然后计算需要给哪些task发消息并建立连接在nimbus将topology终止的时候会将zookeeper上的相关信息删除在集群运行的时候要明白的概念当然消息被传递的时候其实发起者接收者都是而真正执行的是可以理解为一个线程由它来轮询其中的在中通过机制来保证数据至少被处理一次简单来说下在消息发收的过程中会形成一棵树状的结构在一个消息收的时候发一个验证消息发的时候也发一个验证消息那么总体上每个消息出现两次那么机制就是将每个消息的随机生成的进行异或如果在某一时刻结果为那就说明处理成功如下图所示需要补充一下虽然算是随机算法但是出错的概率极低但是系统应该具备在出错之后矫正的能力甚至检查是否出错机制保证了消息会被处理但是不能保证只处理一次顺序处理在需要的情形就有了事务的概念码代码基本用法所谓普通模式是指不去使用为开发人员提供的高级抽象用其提供的原生的接口进行开发主要涉及到的接口有ISpout数据源头接口jstorm会不断调用nextTuple方法来获取数据并发射出去open在worker中初始化该ISpout时调用一般用来设置一些属性比如从spring容器中获取对应的Beanclose和open相对应在要关闭的时候调用activate从非活动状态变为活动状态时调用deactivate和activate相对应从活动状态变为非活动状态时调用nextTupleJStorm希望在每次调用该方法的时候它会通过collectoremit发射一个tupleackjstorm发现msgId对应的tuple被成功地完整消费会调用该方法fail和ack相对应jstorm发现某个tuple在某个环节失败了IBolt数据处理接口jstorm将消息发给他并让其处理完成之后可能整个处理流程就结束了也可能传递给下一个节点继续执行prepare对应ISpout的open方法cleanup对应ISpout的close方法吐槽一下搞成一样的名字会死啊execute处理jstorm发送过来的tupleTopologyBuilder每个jstorm运行的任务都是一个拓扑接口而builder的作用就是根据配置文件构建这个拓扑结构更直白就是构建一个网setSpout添加源头节点并设置并行度setBolt添加处理节点并设置并行度因为还存在多种其他类型的拓扑结构那么在这个环节当然不能乱传在基本用法要去实现接口他们并没有新增任何的方法仅仅是用来区分类型既然是拓扑结构那么应该是一个比较复杂的网络其实这个是在中完成的其中返回的结果其实是对象在其中定义了个流分组的策略字段分组全局分组随机分组本地或随机分组无分组广播分组直接分组自定义分组通过这些接口我们可以一边增加处理节点一边指定其消费哪些消息批量用法基本的用法是每次处理一个但是这种效率比较低很多情况下是可以批量获取消息然后一起处理批量用法对这种方式提供了支持打开代码可以很明显地发现和的有着不小的区别中的定义批次发射成功处理批次中的定义另外如果用批次的话就需要改用来构建拓扑结构在中主要实现的接口如下execute虽然和IBolt中名字参数一致但是增加了一些默认逻辑入参的inputgetValue0表示批次BatchId发送消息时collectoremitnewValuesbatchIdvalue发送的列表第一个字段表示批次BatchIdcommit批次成功时调用常见的是修改offsetrevert批次失败时调用可以在这里根据offset取出批次数据进行重试事务拓扑并不是新的东西只是在原始的上做了一层封装在事务拓扑中以并行和顺序混合的方式来完成任务使用可以保证每个消息只会成功处理一次不过需要注意的是在需要保证能够根据httpsgithubcomalibabajstormblobmasterexamplesequencesplitmergesrcmainjavacomalipaydwjstormtranscationTransactionalGlobalCountjavahttpifevecomgettingstartedofstorm8进行多次重试在这里有一个基本的例子这里有一个不错的讲解这次一种更高级的抽象甚至不需要知道底层是怎么的所面向的不再是和而是主要涉及到下面几种接口在本地完成的操作Function自定义操作Filters自定义过滤partitionAggregate对同批次的数据进行localcombiner操作project只保留stream中指定的fieldstateQuerypartitionPersist查询和持久化决定Tuple如何分发到下一个处理环节shuffle随机broadcast广播partitionBy以某一个特定的field进行hash分到某一个分区这样该field位置相同的都会放到同一个分区global所有tuple发到指定的分区batchGlobal同一批的tuple被放到相同的分区不同批次不同分区partition用户自定义的分区策略不同partition处理结果的汇聚操作aggregate只针对同一批次的数据persistentAggregate针对所有批次进行汇聚并将中间状态持久化对stream中的tuple进行重新分组后续的操作将会对每一个分组独立进行类似sql中的groupbygroupBy将多个Stream融合成一个merge多个流进行简单的合并join多个流按照某个KEY进行UNION操作只能针对同一个批次的数据httpgitlabalibabainccomalohaalohautilityblobmastermetaspoutsrcmainjavacomalibabaalohametaexampleTestTridentTopologyjavaL11在这里有一个中使用的简单例子问题排查模型设计在很多的实际问题中我们面对的模型都是大同小异下面先来看问题是什么问题描述在流式计算中经常需要对一批的数据进行汇总计算如果用来描述就是在用来实现这一条简单的时面对的是一条一条的数据库变化的消息这里需要保证有序消费其实相当于在一堆的消息上面做了一个嵌套的查询用一张图表示如下httpwwwcnblogscomantispamp4182210html出自httpwwwcnblogscomantispamp4182210html