Apache Kafka 实时应用程序(Twitter)

while(i < 10) { Status ret = queue.poll(); if (ret == null) { Thread.sleep(100); i++; }else { for(HashtagEntity hashtage : ret.getHashtagEntities()) { System.out.println("Hashtag: " + hashtage.getText()); producer.send(new ProducerRecord<String, String>( top-icName, Integer.toString(j++), hashtage.getText())); } } } producer.close(); Thread.sleep(5000); twitterStream.shutdown(); }}汇编使用以下命令编译应用程序 - javac -cp "/path/to/kafka/libs/*":"/path/to/twitter4j/lib/*":. KafkaTwitterProducer.java执行打开两个控制台。 在一个控制台中运行上面编译的应用程序，如下所示。java -cp “/path/to/kafka/libs/*":"/path/to/twitter4j/lib/*":. KafkaTwitterProducer <twitter-consumer-key><twitter-consumer-secret><twitter-access-token><twitter-ac-cess-token-secret>my-first-topic food在另一个窗口中运行前一章中解释的Spark / Storm应用程序中的任何一个。 主要要注意的是，在这两种情况下使用的主题应该是相同的。 在这里，我们使用“我的第一主题"作为主题名称。输出此应用程序的输出将取决于关键字和Twitter的当前Feed。 下面指定样本输出(集成storm)。. . .food : 1foodie : 2burger : 1. . .