STORM_0010_Message passing implementation/消息传递的实现

Posted 2020-07-20

下面是0.8.0之前的表述，之后的已经基于Disruptor改造过了

这个文章演示了发射和转移tuple是怎么在storm中工作的

 

• Worker为消息传递负责
• 当zk中的任务出现了变化或者每个task.refresh.poll.secs都会调用refresh-connections。这个东西管理和其他的worker的连接，并且维护一个映射：task->worker
• 提供一个transfer函数，被tasks用来发送tuples给其他的tasks。这个转移函数传入参数是task id和一个tuple，它会序列化一个tuple并且把它放入一个transfer queue。每一个worker只有一个队列。
• 序列化是线程安全的。
• worker使用单线程从队列中取出消息发送给其他的worker。
• 消息发送遵循一定的协议
• 分布式的实现是使用ZeroMQ
• 本地模式的实现是使用内存的java queues
• tasks接收消息的实现在本地模式和分布式模式是不一样的
• 在本地模式，tuple直接被发送到内存队列中
• 在分布式模式中，每一个worker监听一个唯一的tcp端口，等待传入信息，然后路由这些信息从内存到tasks，这些tcp端口称为virtual port,因为它接收[task id,message]，然后路由它到真实的task中。
• tasks监听内存ZeroMQ端口等待从虚拟端口来的消息
• tasks为消息路由负责。一个tuple要么被发送到一个direct stream(这个task id被声明了)，要么去一个regular流。在direct流中，只有bolt订阅了direct stream的时候消息才发送。在regular stream中，流分组函数被使用来决定tuple发送给哪个task id。
• tasks有一个路由对应：{stream id}->{commponent id}->{stream grouping function}
• 无论是direct还是regular，task-fn都返回要发送tuple的task id
• 获取输出的task id之后，bolts和spouts使用worker提供的transfer-fn完成实际的传递tuple的工作

 

 

 

 

Storm Metrics

测量storm

Storm暴露了一个测量接口来报告整个拓扑的统计信息，它在内部使用去追踪在Nimbus UI中看到的数据：excutes和acks的数量，每个bolt的平均处理延时，worker堆的使用，等等

 

Metrics Types

Metrics只实现一个方法，getValueAndRest:做所有的剩余的工作去找到统计信息，并且重置到一个初始状态。例如MapReducer用running total除以running count得到了mean，然后初始化两者的值到0.

Storm提供给你下面的Metrics方法

• AssignableMetric:设置metric为你提供的精确的数值。如果是个外面的值就很方便了，或者在你自己计算出统计信息的情况下。
• CombinedMetric:是测量的通用接口可以同步升级。
• CountMetric:给定值的running total。调用inrc()函数加1，incrBy(n)增加到指定的数值
• MultiCountMetric:CountMetric的hashmap
• ReducedMetric
• MeanReducer:跟踪运行时平均值交给自己的reduce()方法
• MultiMeanReducer:MeanReducer的hashmap