我的书签
添加书签
移除书签Storm 与 Kestrel
本文说明了如何使用 Storm 从 Kestrel 集群中消费数据。
前言
Storm
本教程中使用了 storm-kestrel 项目和 storm-starter 项目中的例子。建议读者将这几个项目 clone 到本地，并动手运行其中的例子。
Kestrel
本文假定读者可以如此项目所述在本地运行一个 Kestrel 集群。
Kestrel 服务器与队列
Kestrel 服务中包含有一组消息队列。Kestrel 队列是一种非常简单的消息队列，可以运行于 JVM 上，并使用 memcache 协议（以及一些扩展）与客户端交互。详情可以参考 storm-kestrel 项目中的 KestrelThriftClient 类的实现。
每个队列均严格遵循先入先出的规则。为了提高服务性能，数据都是缓存在系统内存中的；不过，只有开头的 128MB 是保存在内存中的。在服务停止的时候，队列的状态会保存到一个日志文件中。
请参阅此文了解更多详细信息。
Kestrel 具有 快速 小巧 持久 可靠 等特点。
例如，Twitter 就使用 Kestrel 作为消息系统的核心环节，此文中介绍了相关信息。
** 向 Kestrel 中添加数据
首先，我们需要一个可以向 Kestrel 的队列添加数据的程序。下述方法使用了 storm-kestrel 项目中的 KestrelClient 的实现。该方法从一个包含 5 个句子的数组中随机选择一个句子添加到 Kestrel 的队列中。
1. private static void queueSentenceItems(KestrelClient kestrelClient, String queueName)
2. throws ParseError, IOException {
4. String[] sentences = new String[] {
5. "the cow jumped over the moon",
6. "an apple a day keeps the doctor away",
7. "four score and seven years ago",
8. "snow white and the seven dwarfs",
9. "i am at two with nature"};
11. Random _rand = new Random();
13. for(int i=1; i<=10; i++){
15. String sentence = sentences[_rand.nextInt(sentences.length)];
17. String val = "ID " + i + " " + sentence;
19. boolean queueSucess = kestrelClient.queue(queueName, val);
21. System.out.println("queueSucess=" +queueSucess+ " [" + val +"]");
22. }
23. }
从 Kestrel 中移除数据
此方法从一个队列中取出一个数据，但并不把该数据从队列中删除：
1. private static void dequeueItems(KestrelClient kestrelClient, String queueName) throws IOException, ParseError { for(int i=1; i<=12; i++){
3. Item item = kestrelClient.dequeue(queueName);
5. if(item==null){
6. System.out.println("The queue (" + queueName + ") contains no items.");
7. }
8. else
9. {
10. byte[] data = item._data;
12. String receivedVal = new String(data);
14. System.out.println("receivedItem=" + receivedVal);
15. }
16. }
此方法会从队列中取出并移除数据：
1. private static void dequeueAndRemoveItems(KestrelClient kestrelClient, String queueName)
2. throws IOException, ParseError
3. {
4. for(int i=1; i<=12; i++){
6. Item item = kestrelClient.dequeue(queueName);
9. if(item==null){
10. System.out.println("The queue (" + queueName + ") contains no items.");
11. }
12. else
13. {
14. int itemID = item._id;
17. byte[] data = item._data;
19. String receivedVal = new String(data);
21. kestrelClient.ack(queueName, itemID);
23. System.out.println("receivedItem=" + receivedVal);
24. }
25. }
26. }
向 Kestrel 中连续添加数据
下面的程序可以向本地 Kestrel 服务的一个 sentence_queue 队列中连续添加句子，这也是我们的最后一个程序。
可以在命令行窗口中输入一个右中括号 ] 并回车来停止程序。
1. import java.io.IOException;
2. import java.io.InputStream;
3. import java.util.Random;
5. import backtype.storm.spout.KestrelClient;
6. import backtype.storm.spout.KestrelClient.Item;
7. import backtype.storm.spout.KestrelClient.ParseError;
9. public class AddSentenceItemsToKestrel {
11. /**
12. * @param args
13. */
14. public static void main(String[] args) {
16. InputStream is = System.in;
18. char closing_bracket = ']';
20. int val = closing_bracket;
22. boolean aux = true;
24. try {
26. KestrelClient kestrelClient = null;
27. String queueName = "sentence_queue";
29. while(aux){
31. kestrelClient = new KestrelClient("localhost",22133);
33. queueSentenceItems(kestrelClient, queueName);
35. kestrelClient.close();
37. Thread.sleep(1000);
39. if(is.available()>0){
40. if(val==is.read())
41. aux=false;
42. }
43. }
44. } catch (IOException e) {
45. // TODO Auto-generated catch block
46. e.printStackTrace();
47. }
48. catch (ParseError e) {
49. // TODO Auto-generated catch block
50. e.printStackTrace();
51. } catch (InterruptedException e) {
52. // TODO Auto-generated catch block
53. e.printStackTrace();
54. }
56. System.out.println("end");
58. }
59. }
使用 KestrelSpout
下面的拓扑使用 KestrelSpout 从一个 Kestrel 队列中读取句子，并将句子分割成若干个单词（Bolt：SplitSentence），然后输出每个单词出现的次数（Bolt：WordCount）。数据处理的细节可以参考消息的可靠性保证一文。
1. TopologyBuilder builder = new TopologyBuilder();
2. builder.setSpout("sentences", new KestrelSpout("localhost",22133,"sentence_queue",new StringScheme()));
3. builder.setBolt("split", new SplitSentence(), 10)
4. .shuffleGrouping("sentences");
5. builder.setBolt("count", new WordCount(), 20)
6. .fieldsGrouping("split", new Fields("word"));
运行
首先，以生产模式或者开发者模式启动你的本地 Kestrel 服务。
然后，等待大约 5 秒钟以防出现网络连接异常。
现在可以运行向队列中添加数据的程序，并启动 Storm 拓扑。程序启动的顺序并不重要。
如果你以 TOPOLOGY_DEBUG 模式运行拓扑你会观察到拓扑中 tuple 发送的细节信息。
