内容简介:在某些情况下,Bolt 在执行某些操作之前需要将数据缓存几秒钟,例如每隔5秒清理一次缓存或在单个请求中将一批记录插入数据库。Tick 元组是系统生成的(Storm生成的)元组,我们可以在每个 Bolt 级别配置它们。我们可以在编写 Bolt 时在代码中配置 Tick 元组。我们需要在 Bolt 中覆盖以下方法以启用 Tick 元组:
在某些情况下,Bolt 在执行某些操作之前需要将数据缓存几秒钟,例如每隔5秒清理一次缓存或在单个请求中将一批记录插入数据库。
Tick 元组是系统生成的(Storm生成的)元组,我们可以在每个 Bolt 级别配置它们。我们可以在编写 Bolt 时在代码中配置 Tick 元组。
我们需要在 Bolt 中覆盖以下方法以启用 Tick 元组:
@Override public Map<String, Object> getComponentConfiguration() { Config conf = new Config(); int tickFrequencyInSeconds = 10; conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, tickFrequencyInSeconds); return conf; }
在上面的代码中,我们将 Tick 元组配置为10秒。现在,Storm 会每10秒钟生成一个 Tick 元组。
接下来创建 isTickTuple 方法来确定我们收到的元组是 Tick 元组还是正常元组:
protected static boolean isTickTuple(Tuple tuple) { return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID); }
Tick 元组会与你正在处理的其他正常元组混合在一起,所以需要我们判断元组的类型。
最后,在 Bolt 的 execute 方法中添加如下代码来判断元组的类型进行处理:
@Override public void execute(Tuple tuple, BasicOutputCollector collector) { try { if (isTickTuple(tuple)) { // do tick tuple } else { // do normal tuple } // do your bolt stuff } catch (Exception e) { LOG.error("Bolt execute error: {}", e); collector.reportError(e); } }
现在你的 Bolt 每10秒就会收到一个 Tick 元组。
如果希望 Topology 中的每个 Bolt 都每隔一段时间做一些操作,那么可以定义一个 Topology 全局的 Tick,同样是设置 Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS
的值:
Config conf = new Config(); conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
创业的艺术2.0
〔美〕盖伊·川崎 / 刘悦、段歆玥 / 译言·东西文库/电子工业出版社 / 2016-9 / 68
“创业者导师”——盖伊•川崎的《创业的艺术2.0》被阿丽亚娜•赫芬顿评为“终极的创业手册”。无论您是企业家、小企业主、企业开拓者还是非盈利组织的领导人,都可以让你的产品、服务或理念获得成功。 盖伊选取了不用角度,探索前十年商界的巨大变化,并寻求解决之道。曾经所向披靡的市场巨头深陷水深火热之中,社交媒体也取代了人际关系和广告,成为营销推广的主要渠道。众筹也成为广大投资者的可行之举。“云”更是每......一起来看看 《创业的艺术2.0》 这本书的介绍吧!