博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
理解Storm可靠性消息
阅读量:4582 次
发布时间:2019-06-09

本文共 2187 字,大约阅读时间需要 7 分钟。

看过一些别人写的, 感觉有些东西没太说清楚,个人主要以源代码跟踪,参考个人理解讲述,有错误请指正。

1基本名词

1.1 Tuple: 消息传递的基本单位。很多文章中介绍都是这么说的, 个人觉得应该更详细一点。

 在spout发送的时候,函数原型

 public List<Integer> emit(List<Object> tuple, Object messageId) {

        return emit(Utils.DEFAULT_STREAM_ID, tuple, messageId);
    }

这里的tuple, 实际上是List<Object> 对象,返回的是 List<Integer> 是要发送的tast的IdsList

在bolt接收的时候, 函数原型

public void execute(Tuple tuple)

变成了一个Tuple对象,  结构应该也是一个list, List<Field1, value1, Field2, value2..>这样的一个结构, FieldList ValueList, 我们根据对应的fieldname就可以取出对应的getIntegerByField方法

回到spout对象中来, 在spout有一个定义的输出字段

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        declarer.declare(new Fields("word"));
    }

这里定义的一个字段,所以我们在emit的时候就只能发送一个包含一个value的tuple(spout部分), storm会将field, 和 发送的value下标对应, 变成一个Tuple对象,  也就是上面说的

List<Field1, value1, Field2, value2..>这样的一个结构,  在bolt 之间传递tuple, 发送又是List<Object> tuple, 根据组装bolt定义的fiels, 再组合成Tuple对象给下一个Bolt处理

在发射的最后 还有一个 void emitDirect(int taskId, String streamId, List<Object> tuple, Object messageId);  因为上面emit的时候已经返回List<taskid>, 所以它就知道要发送给哪些taskid处理,然后将taskid 和 tuple放入队列LinkedBlockingQueue, 代码如下

; worker.clj

( defn mk-transfer- fn [ transfer-queue ]

( fn [ task ^Tuple tuple ]

(.put ^LinkedBlockingQueue

transfer-queue [ task tuple ] )

))
然后单独会开启一个叫async-loop的线程,取出每条记录(taskid, tuple), 然后worker会从当前task建立一个到目标task的zeromq连接, 通过zeromq将tuple发送给目标task
 

总结: 每次emit都是根据List<Object>和定义的输出Fields组合成一个Tuple对象,,每个接受对象接收的是Tuple对象,如果处理完再发送又再组合字段, 在emit的时候返回LIst<taskids>,所以就知道发送给哪些Task, 然后拿这些taskid和tuple再组合成一个任务队列,通过zeromq发送到目标task,目标task接收到tuple进程处理至于并发度控制, 参考

http://www.cnblogs.com/chengxin1982/p/4001275.html

 

TupleID Tuple对应的ID,  在创建的时候赋予一个64位的id,主要用来跟踪消息

MsgID  官方解释 Emits a new tuple to the default output stream with the given message ID. 如果不指定,acker不会跟踪。主要作用 , 在spout收到fail时候, 能够定位到是哪条消息出错,能够决定重发. 使用实例  _collector.emit(new Values(sentence),  new Integer(num));

acker 消息跟踪者. acker 存储一个Map<taskid, ack val> ,  taskid为祖宗tuple创建者的taskid ack_val 为消息传递过程中的 tupleid的xor值,如果为0则知道是哪个spout或者bolt已经处理完了, 为什么会有bolt, 因为bolt在发射的时候,如果非锚定,就是不带tuple发射,它会被认为是祖宗tuple, 上一个tuple会认为已经结束.

至于分配发射源分配到acker, storm采用一致性hash 祖宗tupleid来分配,因为在所有的tuple中都能知道祖宗tupleid,所以在子孙tuple处理时, 知道该发送给哪个acker跟踪

 

转载于:https://www.cnblogs.com/chengxin1982/p/4004826.html

你可能感兴趣的文章
[洛谷P5057][CQOI2006]简单题
查看>>
多线程同步的几种方法
查看>>
数据结构-冒泡排序
查看>>
关于程序状态字寄存器PSW(Program Status Word)与多核多线程
查看>>
mybatis的缓存
查看>>
java 缓冲流 Buffer
查看>>
7月23号=》261页-265页
查看>>
软考知识点梳理--综合布线
查看>>
Mysql5.6主从热备配置
查看>>
VS2010DebugView捕捉
查看>>
mfix中更改time dependent VTK filename的最大时间步数的容量
查看>>
Windows7安装 docker-compose的过程
查看>>
关于nodeJS多线程的支持,目前看来无法实现,讲解v8的一些东西
查看>>
php递归创建文件夹的两种方法
查看>>
6.新增事件
查看>>
|洛谷|二分|P1182 数列分段Section II
查看>>
少儿编程Scratch第四讲:射击游戏的制作,克隆的奥秘
查看>>
Oracle学习第七课-表连接及其应用
查看>>
Python基础篇【第十三篇】:面向对象
查看>>
bzoj 2465 小球
查看>>