版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/zhangkezhi_471885889/article/details/71218895
背景
由于cleanup方法并不可靠,它只在local mode下生效,Storm集群模式下cleanup不会被调用执行。很多资源得不到释放解决方案
在kill topology之前,先deactivate相应的topology。在spout中实现deactivate()方法,deactivate()方法中给bolt emit特殊的数据(如:emit “shutDown”字符串给bolt),bolt中判断接收的数据为”shutDown”就调用cleanup()方法。在cleanup()方法中释放需要释放的资源。流程图
相关代码
spout重写deactivate()方法代码如下:
@Override
public void deactivate(){
LOGGER.info("deaactivate to spout and bolt");
try {
//storm deactivate时发一条消息给bolt
collector.emit(new Values("shutDown"));
} catch (Exception e) {
e.printStackTrace();
}
}
bolt中execute()方法代码如下:
@Override
public void execute(Tuple input) {
String message = input.getStringByField("loan_message");
//判断是不是spout的deactivate传过来的消息
if("shutDown".equals(message )) {
cleanup();
}
}
注意事项: deactivate topology时,建议等待时间尽量长,时间过短消息来不及处理,会导致数据丢失