Storm集群模式下cleanup解决方法

xiaoxiao2021-02-28  64

背景 由于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时,建议等待时间尽量长,时间过短消息来不及处理,会导致数据丢失

转载请注明原文地址: https://www.6miu.com/read-35897.html

最新回复(0)