创建一个Apache Storm spout，它每X秒发出一个元组

Question

>

每x秒发出一批元组（或单个元组中的字符串列表）。我如何实现这一点？我读了一些关于Storm Trident的文章，但它的ibatchspout似乎不允许我以特定的时间间隔批量发出元组。

如果没有新的数据进来，喷口应该怎么做？它不能阻挡线程，因为它是Storm的主线程，对吧？

Answer 1

您可以实现自己的MQTT spout。例如，请查看MongoSpout。

重要的部分是NextTuple方法。

调用此方法时，Storm请求Spout向输出收集器发出元组。这个方法应该是非阻塞的，所以如果喷口没有要发出的元组，这个方法应该返回。nextTuple、ack和fail都是在spout任务的单个线程中的紧密循环中调用的。当没有元组要发出时，有礼貌的做法是让nextTupleHibernate一小段时间（比如一毫秒），这样就不会浪费太多CPU。

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);
}