小背景:我正在使用ApacheStorm开发一个拓扑,我想为什么不在其中使用依赖注入,但我不确定当拓扑部署到集群时它在集群环境中的行为。我开始寻找di是否适合在storm拓扑中使用的答案,我遇到了一些关于ApacheSpark的线程,其中提到序列化将是一个问题,并看到了ApacheStorm的一些响应。最后,我决定用googleguice编写一个示例拓扑,看看会发生什么。
我用两个螺栓编写了一个示例拓扑,并使用googleguice注入依赖项。第一个bolt发出一个tick元组,然后第一个bolt创建消息,bolt在日志上打印消息并调用一些类,这些类执行相同的操作。然后,这个消息被发送到第二个bolt,同样的打印逻辑也在那里。
第一个螺栓
public class FirstBolt extends BaseRichBolt {
private OutputCollector collector;
private static int count = 0;
private FirstInjectClass firstInjectClass;
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
collector = outputCollector;
Injector injector = Guice.createInjector(new Module());
firstInjectClass = injector.getInstance(FirstInjectClass.class);
}
@Override
public void execute(Tuple tuple) {
count++;
String message = "Message count "+count;
firstInjectClass.printMessage(message);
log.error(message);
collector.emit("TO_SECOND_BOLT", new Values(message));
collector.ack(tuple);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declareStream("TO_SECOND_BOLT", new Fields("MESSAGE"));
}
@Override
public Map<String, Object> getComponentConfiguration() {
Config conf = new Config();
conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);
return conf;
}
}
第二个螺栓
public class SecondBolt extends BaseRichBolt {
private OutputCollector collector;
private SecondInjectClass secondInjectClass;
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
collector = outputCollector;
Injector injector = Guice.createInjector(new Module());
secondInjectClass = injector.getInstance(SecondInjectClass.class);
}
@Override
public void execute(Tuple tuple) {
String message = (String) tuple.getValue(0);
secondInjectClass.printMessage(message);
log.error("SecondBolt {}",message);
collector.ack(tuple);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
}
}
注入依赖项的类
public class FirstInjectClass {
FirstInterface firstInterface;
private final String prepend = "FirstInjectClass";
@Inject
public FirstInjectClass(FirstInterface firstInterface) {
this.firstInterface = firstInterface;
}
public void printMessage(String message){
log.error("{} {}", prepend, message);
firstInterface.printMethod(message);
}
}
用于绑定的接口
public interface FirstInterface {
void printMethod(String message);
}
接口的实现
public class FirstInterfaceImpl implements FirstInterface{
private final String prepend = "FirstInterfaceImpl";
public void printMethod(String message){
log.error("{} {}", prepend, message);
}
}
与另一个通过di接收依赖关系的类相同
public class SecondInjectClass {
SecondInterface secondInterface;
private final String prepend = "SecondInjectClass";
@Inject
public SecondInjectClass(SecondInterface secondInterface) {
this.secondInterface = secondInterface;
}
public void printMessage(String message){
log.error("{} {}", prepend, message);
secondInterface.printMethod(message);
}
}
另一个绑定接口
public interface SecondInterface {
void printMethod(String message);
}
第二接口的实现
public class SecondInterfaceImpl implements SecondInterface{
private final String prepend = "SecondInterfaceImpl";
public void printMethod(String message){
log.error("{} {}", prepend, message);
}
}
模块类
public class Module extends AbstractModule {
@Override
protected void configure() {
bind(FirstInterface.class).to(FirstInterfaceImpl.class);
bind(SecondInterface.class).to(SecondInterfaceImpl.class);
}
}
这里没什么特别的,只有两个螺栓和两个di类。我把它部署在服务器上,效果很好。但问题是我必须初始化 Injector
在每一个螺栓,这让我怀疑它会有什么副作用?
这个实现很简单,只需两个螺栓。。如果我有更多的螺栓呢?如果我必须初始化所有螺栓中的喷油器,它会对拓扑产生什么影响?
如果尝试在prepare方法之外初始化注入器,则序列化时会出错。
暂无答案!
目前还没有任何答案,快来回答吧!