创建一个，它每隔X秒发射一次元组

Question

我有一个从MQTT代理接收数据的拓扑结构，我希望有这样的一个喷口：

1. 每x秒发出一批元组(或单个元组中的字符串列表)。我怎样才能做到这一点？我读过一些关于风暴三叉戟的文章，但是它的IBatchSpout似乎不允许我在一个特定的时间间隔内批量地发出元组。
2. 如果没有新的数据输入，喷口应该做什么？它不能阻止线程，因为它是Storm的主线程，对吗？

Answer 1

您可以实现您自己的MQTT喷口。例如，请看一下MongoSpout。

重要的部分是nextTuple方法。

调用此方法时，Storm将请求Spout向输出收集器发出元组。这个方法应该是非阻塞的，所以如果Spout没有要发出的元组，这个方法应该返回。nextTuple、ack和fail都是在单线程中的一个紧循环中调用的。当没有元组可以发出时，让nextTuple睡眠一小段时间(比如一毫秒)是礼貌的，以免浪费太多的CPU。

您不能一次等待指定的时间，但是您可以实现nextTuple，以便它只偶尔发出一个元组。

private static final EMISSION_PERIOD = 2000; // 2 seconds
private long lastEmission;

@Override
public void nextTuple() {
    if (lastEmission == null ||
            lastEmission + EMISSION_PERIOD >= System.currentMillis()) {
        List<Object> tuple = pollMQTT();
        if (tuple != null) {
            this.collector.emit(tuple);
            return;
        }
    }
    Utils.sleep(50);
}

请注意，我找到了一个开源MQTT喷口。它看上去还没有准备好生产，但您可以使用它作为起点。

Answer 2

除了Christian，我还为Storm的MQTT客户端找到了这一实现。前面提到的链接仍未开发。