java - vertx.io java + twitter4j - 事件循环集成

Question

我正在尝试将 twitter4j 与 vertx 事件循环集成，但我不知道我的做法是否正确。尽管我是一名 Node.js 开发人员，但我还是 vertx 的新手，因此我熟悉事件循环/单线程概念。

在我的测试中，我想订阅 Twitter 流并在 vertx 事件总线上发布该流。

我创建了一个不可知的 TwitterAPI 类，它将把 twitter 流转换为 Observable(稍后将挂接到 vertx):

public class TwitterAPI {
    public Observable<Status> getTwitterObservable() {
        return Observable.create(emitter -> {
            final TwitterStream twitterStream = new TwitterStreamFactory().getInstance();
            twitterStream.addListener(new StatusListener(){
                public void onStatus(Status status) {
                    emitter.onNext(status);
                }
                public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {}
                public void onTrackLimitationNotice(int numberOfLimitedStatuses) {}
                public void onException(Exception ex) {
                    emitter.onError(ex);
                }
                @Override
                public void onScrubGeo(long userId, long upToStatusId) {}

                @Override
                public void onStallWarning(StallWarning warning) {}
            });

            //twitterStream.filter("some keyword");
            twitterStream.sample();
        });
    }
}

然后我创建了一个 TwitterVerticle，它将监听上面的 Observable 并将流发布到事件总线上，以便其他一些 verticle 可以订阅它并处理它:

public class TwitterVerticle extends AbstractVerticle {
    public void start() {
        EventBus eb = this.vertx.eventBus();
        TwitterAPI twitterAPI = new TwitterAPI();
        twitterAPI.getTwitterObservable()
            .map(Status::getText)
            .filter(text -> text.startsWith("my keyword"))
            .subscribe(text -> {
                eb.publish("tweet-feed", text);
            });
    }
}

例如，我创建了另一个 verticle，它将监听事件总线上的“twitter-feed”并将其发布到 WebSocket 上，以便您可以在浏览器中看到提要。

乍一看一切都运行良好...但是...我的主要问题是:我不确定 twitter4j 能否很好地处理事件循环，也许我的集成技术是错误的。也许我应该让 TwitterVerticle 成为 Worker Verticle？

有人可以看一下并告诉我这是否是完成此类任务的最佳方法吗？

非常感谢!

编辑

在这种情况下，直接在事件总线上发布是更好的模式吗？下面是修改后的代码:

public class TwitterAPI {

    public void publishStreamOnEventBus(Vertx vertx) {
        EventBus eb = vertx.eventBus();
        TwitterStream twitterStream = new TwitterStreamFactory().getInstance();

        twitterStream.addListener(new StatusListener(){
            public void onStatus(Status status) {
                eb.publish("tweet-feed", status.getText());
            }
            public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {}
            public void onTrackLimitationNotice(int numberOfLimitedStatuses) {}
            public void onException(Exception ex) {
                //emitter.onError(ex);
            }
            @Override
            public void onScrubGeo(long userId, long upToStatusId) {}

            @Override
            public void onStallWarning(StallWarning warning) {}
        });

        twitterStream.sample();
    }
}



public class Main {

    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        new TwitterAPI().publishStreamOnEventBus(vertx);

        //vertx.deployVerticle(new TwitterVerticle()/*, new DeploymentOptions().setWorker(true)*/);
        vertx.deployVerticle(new WebServerVerticle());
    }

}

Answer 1

首先是一些一般性建议。

1/如果订阅 Observable 涉及调用阻塞 API，请使用 blockingScheduler:

Scheduler blockingScheduler = io.vertx.rx.java.RxHelper.blockingScheduler(vertx);
Observable<String> obs = twitterObservable.subscribeOn(blockingScheduler);

2/然后我假设 twitter4j 使用自己的线程来调用 StatusListener，因此 EventBus#publish 调用将在其中一个上进行那些。要返回到 verticle 上下文，请使用 #observeOn 运算符:

Scheduler contextScheduler = io.vertx.rx.java.RxHelper.scheduler(context);
Observable<String> obs = twitterObservable.observeOn(contextScheduler);

结合这两个更改:

public class TwitterVerticle extends AbstractVerticle {
    public void start() {
        EventBus eb = this.vertx.eventBus();
        Scheduler contextScheduler = io.vertx.rx.java.RxHelper.scheduler(context);
        Scheduler blockingScheduler = io.vertx.rx.java.RxHelper.blockingScheduler(vertx);
        TwitterAPI twitterAPI = new TwitterAPI();
        twitterAPI.getTwitterObservable()
            .map(Status::getText)
            .filter(text -> text.startsWith("my keyword"))
            .observeOn(contextScheduler)
            .subscribeOn(blockingScheduler)
            .subscribe(text -> {
                eb.publish("tweet-feed", text);
            });
    }
}

综上所述，如果这是 verticle 的唯一工作，我建议摆脱它并简单地发布到事件总线。事件总线实例是线程安全的，并且完全可以从外部(非 Vert.x)世界调用 publish。实际上，这是将 Vert.x 代码与遗留代码结合起来的一个很好的模式。