通过 Twitter4J 进行多线程 Twitter 访问答案

【问题标题】：Multithreaded Twitter access through Twitter4J通过 Twitter4J 进行多线程 Twitter 访问
【发布时间】：2014-09-02 06:13:11
【问题描述】：

我编写了以下 Java 代码：

twitterStream.addListener(new StreamListener());

FilterQuery filterQuery = new FilterQuery();
filterQuery.follow(filteringUsers);
filterQuery.track(filteringWords);

twitterStream.filter(filterQuery);

跟踪 Twitter 中的一些用户和关键字（通过 Streaming API）。这里StreamListener是我个人实现的监听器。

我正在跟踪很多关键字、主题标签和用户，因此我在内存中积累了大量等待处理的推文。事实上，我只是将它们通过监听器（在onStatus() 方法中）并将它们刷新到数据库中。

不过，他们必须在内存中等待的事实显然会在几个小时内使内存饱和。在 20 分钟的运行中，我在内存中积累了 177000 个LinkedBlockingQueue$Node 对象和 1.272MB 的char[]（通过分析查看）。

我想保持管道连续运行，显然这在当前状态下是不可能的。

所以想知道有没有办法在多线程中添加多个监听器，让他们可以同时清空推文队列，加快处理速度。

如果可能：这些侦听器是否同时清空队列？我的意思是：他们是否会多次阅读同一条推文？
如果不可能：我该如何解决我的问题？

提前致谢。

【问题讨论】：

标签： java multithreading twitter twitter4j

【解决方案1】：

虽然通过 Twitter4J 无法实现直接的多线程解决方案，但可以决定通过侦听器类模拟多线程队列处理。

假设StreamListener 是您对StatusListener Twitter4J 监听器的专业化。

我们将StreamListener 中的队列复制为私有属性：

private LinkedBlockingQueue<String> tweets;

队列在构造函数中初始化：

tweets = new LinkedBlockingQueue<String>();

此外，在构造函数中，我们构建了一个线程池，用于从队列中（批量）读取推文并将它们存储在数据库中：

    final ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
    Runnable tweetAnalyzer = defineMonitoringRunnable(tweetRepository);
    for (int i = 0; i < NUM_THREADS; i++) {
        executor.execute(tweetAnalyzer);
        try {
            Thread.sleep(THREADS_DELAY);
        } 
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

Runnable 对象的构建方式如下：

private Runnable defineMonitoringRunnable(final TweetRepository tweetRepository) {
    return new Runnable() {

        @Override
        public void run() {
            List<String> tempTweets = new ArrayList<String>();

            while (true) {
                if (tweets.size() > 0) {
                    tempTweets.clear();
                    tweets.drainTo(tempTweets);

                    tweetRepository.insert(tempTweets);   
                }

                try {
                    Thread.sleep(TWEETS_SAVING_TIME);
                } 
                catch (InterruptedException ex) {
                    ex.printStackTrace();
                }

            }
        }
    };
}

（TWEETS_SAVING_TIME 是每个Thread 对象在一条推文保存和另一条推文之间的等待时间）

最后，onStatus() 方法在推文到达侦听器后将其存储在队列中：

@Override
public void onStatus(Status status) {   
    tweets.add(TwitterObjectFactory.getRawJSON(status));
}

【讨论】：