对的,这是可能的:
TopologyBuilder b = new TopologyBuilder();
b.setSpout("topic_1", new KafkaSpout(...));
b.setSpout("topic_2", new KafkaSpout(...));
b.setBolt("bolt", new MyBolt(...)).shuffleGrouping("topic_1").shuffleGrouping("topic_2");
您也可以使用任何其他分组。
Update:
为了区分消费者bolt中的元组(即topic_1或topic_2),有两种可能性:
1) 您可以使用操作员 ID(如 @user-4870385 建议):
if(input.getSourceComponent().equalsIgnoreCase("topic_1")) {
//do something
} else {
//do something
}
2)您可以使用流名称(按照@zenbeni的建议)。对于这种情况,两个 Spout 都需要声明命名流,并且 Bolt 需要通过流名称连接到 Spout:
public class MyKafkaSpout extends KafkaSpout {
final String streamName;
public MyKafkaSpout(String stream) {
this.streamName = stream;
}
// other stuff omitted
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// compare KafkaSpout.declareOutputFields(...)
declarer.declare(streamName, _spoutConfig.scheme.getOutputFields());
}
}
构建拓扑,现在需要使用流名称:
TopologyBuilder b = new TopologyBuilder();
b.setSpout("topic_1", new MyKafkaSpout("stream_t1"));
b.setSpout("topic_2", new MyKafkaSpout("stream_t2"));
b.setBolt("bolt", new MyBolt(...)).shuffleGrouping("topic_1", "stream_t1").shuffleGrouping("topic_2", "stream_t2");
In MyBolt
流名称现在可用于区分输入元组:
// in my MyBolt.execute():
if(input.getSourceStreamId().equals("Topic1")) {
// do something
} else {
// do something
}
讨论:
虽然second使用流名称的方法更自然(根据@zenbeni),first更灵活(IHMO)。流名称由spout/bolt直接声明(即在编写spout/bolt代码时);相反,操作员 ID 是在拓扑组合在一起时分配的(即,在喷口/螺栓被放置时)used).
假设我们获得三个螺栓作为类文件(没有源代码)。前两个应该用作生产者,并且都声明具有相同名称的输出流。如果第三个消费者通过流来区分输入元组,则这将不起作用。即使两个给定的生产者 Bolt 声明不同的输出流名称,预期的输入流名称也可能会硬编码在消费者 Bolt 中,并且可能不匹配。因此,它也不起作用。但是,如果消费者 Bolt 使用组件名称(即使它们是硬编码的)来区分传入的元组,则可以正确分配预期的组件 ID。
当然,可以从给定的类继承(如果没有声明final
并覆盖declareOutputFields(...)
为了分配自己的流名称。然而,这是更多额外的工作要做。