storm的ack消息不丢失机制

[复制链接]
查看: 4650|回复: 0

34

主题

38

帖子

489

积分

管理员

Rank: 9Rank: 9Rank: 9

积分
489
发表于 2019-11-21 22:12:25 | 显示全部楼层 |阅读模式
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/weixin_42295141/article/details/82746254


1:ack是什么

ack 机制是storm整个技术体系中非常闪亮的一个创新点。

通过Ack机制,spout发送出去的每一条消息,都可以确定是被成功处理或失败处理, 从而可以让开发者采取动作。比如在Meta中,成功被处理,即可更新偏移量,当失败时,重复发送数据。

因此,通过Ack机制,很容易做到保证所有数据均被处理,一条都不漏。

另外需要注意的,当spout触发fail动作时,不会自动重发失败的tuple,需要spout自己重新获取数据,手动重新再发送一次

ack机制即, spout发送的每一条消息,

在规定的时间内,spout收到Acker的ack响应,即认为该tuple 被后续bolt成功处理
在规定的时间内,没有收到Acker的ack响应tuple,就触发fail动作,即认为该tuple处理失败,
或者收到Acker发送的fail响应tuple,也认为失败,触发fail动作
另外Ack机制还常用于限流作用: 为了避免spout发送数据太快,而bolt处理太慢,常常设置pending数,当spout有等于或超过pending数的tuple没有收到ack或fail响应时,跳过执行nextTuple, 从而限制spout发送数据。

通过conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, pending);设置spout pend数。

这个timeout时间可以通过Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS来设定。Timeout的默认时长为30秒

2:如何使用Ack机制

spout 在发送数据的时候带上msgid

设置acker数至少大于0;Config.setNumAckers(conf, ackerParal);

在bolt中完成处理tuple时,执行OutputCollector.ack(tuple), 当失败处理时,执行OutputCollector.fail(tuple);

推荐使用IBasicBolt, 因为IBasicBolt 自动封装了OutputCollector.ack(tuple), 处理失败时,抛出FailedException,则自动执行OutputCollector.fail(tuple)

3:如何关闭Ack机制

有2种途径

spout发送数据是不带上msgid

设置acker数等于0

4:基本实现

Storm 系统中有一组叫做"acker"的特殊的任务,它们负责跟踪DAG(有向无环图)中的每个消息。

acker任务保存了spout id到一对值的映射。第一个值就是spout的任务id,通过这个id,acker就知道消息处理完成时该通知哪个spout任务。第二个值是一个64bit的数字,我们称之为"ack val", 它是树中所有消息的随机id的异或计算结果。

<TaskId,<RootId,ackValue>>

Spoutid,<系统生成的id,ackValue>

Task-0,64bit,0

ack val表示了整棵树的的状态,无论这棵树多大,只需要这个固定大小的数字就可以跟踪整棵树。当消息被创建和被应答的时候都会有相同的消息id发送过来做异或。 每当acker发现一棵树的ack val值为0的时候,它就知道这棵树已经被完全处理了








本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有帐号?立即注册

x
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

精选推荐

返回顶部 关注微信 下载APP 返回列表