本文所述程序实例主要实现在后端应用服务器上实时获取STORM集群的运行信息和topology相关的提交和控制。对此，通过对STORM UI和CMD源码的分析，得出可以通过其thrift接口调用实现这些功能。先下载一个thrift库进行编码和安装。关于thrift可以参见这个地方。安装完成后，从STORM源码中将storm.thrift拷贝到thrift目录下。输入:
|
1
|
hrift -gen cpp storm.thrift |
会得到一个gen-cpp目录，里面就是thrift先关脚本的C++实现。我们先看storm.thrift文件接口：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
service Nimbus { //TOPOLOGY上传接口 void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology); void submitTopologyWithOpts(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology, 5: SubmitOptions options); void killTopology(1: string name); void killTopologyWithOpts(1: string name, 2: KillOptions options) throws (1: NotAliveException e); void activate(1: string name) ; void deactivate(1: string name); void rebalance(1: string name, 2: RebalanceOptions options); //TOPOLOGY JAR包上传接口 string beginFileUpload(); void uploadChunk(1: string location, 2: binary chunk); void finishFileUpload(1: string location); string beginFileDownload(1: string file); binary downloadChunk(1: string id); //获取NIMBUS的配置信息 string getNimbusConf(); //获取STORM集群运行信息 ClusterSummary getClusterInfo(); //获取TOPOLOGY的运行状态信息 TopologyInfo getTopologyInfo(1: string id); //获取TOPOLOGY对象信息 string getTopologyConf(1: string id); StormTopology getTopology(1: string id); StormTopology getUserTopology(1: string id);} |
生成C++文件后，我们就可以对其接口进行调用，由于thrift c++框架是使用boost库实现的，必须安装boost库依赖。实现的代码如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
#define HAVE_NETDB_H //使用网络模块的宏必须打开#include "Nimbus.h"#include "storm_types.h"#include <string>#include <iostream>#include <set>#include <transport/TSocket.h> #include <transport/TBufferTransports.h> #include <protocol/TBinaryProtocol.h> int test_storm_thrift(){ boost::shared_ptr<TSocket> tsocket(new TSocket("storm-nimbus-server", 6627)); boost::shared_ptr<TTransport> ttransport(new TFramedTransport(tsocket, 1024 * 512)); //此处必须使用TFramedTransport boost::shared_ptr<TProtocol> tprotocol(new TBinaryProtocol(ttransport)); try{ //创建一个nimbus客户端对象 NimbusClient client(tprotocol); //打开通道 ttransport->open(); ClusterSummary summ; std::string conf; //对STORM的RPC调用，直接获取信息，同步进行的。 client.getNimbusConf(conf); client.getClusterInfo(summ); //关闭通道 ttransport->close(); }catch(TException &tx){ printf("InvalidOperation: %s", tx.what()); }} |
以上代码就可以直接获取nimbus的配置和集群信息，其他接口以此类推。值得注意的是storm.thrift to C++生成的storm_types.h文件里其中operator < 函数都未实现，所以必须手动进行添加实现，否则编译会有问题。
此外，不仅仅C++可以实现STORM的控制，PHP和其他的语言也可以实现，只要thrift支持就OK。感兴趣的读者可以自己实现一下试试看。
