1 IDL
首先是storm.thrift, 作為IDL里面定義了用到的數據結構和service
然后backtype.storm.generated, 存放從IDL通過Thrift自動轉化成的Java代碼
比如對於nimbus service
在IDL的定義為,
service Nimbus {
void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology) throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite);
void submitTopologyWithOpts(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology, 5: SubmitOptions options) throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite);
void killTopology(1: string name) throws (1: NotAliveException e);
void killTopologyWithOpts(1: string name, 2: KillOptions options) throws (1: NotAliveException e);
void activate(1: string name) throws (1: NotAliveException e);
void deactivate(1: string name) throws (1: NotAliveException e);
void rebalance(1: string name, 2: RebalanceOptions options) throws (1: NotAliveException e, 2: InvalidTopologyException ite);
// need to add functions for asking about status of storms, what nodes they're running on, looking at task logs
string beginFileUpload();
void uploadChunk(1: string location, 2: binary chunk);
void finishFileUpload(1: string location);
string beginFileDownload(1: string file);
//can stop downloading chunks when receive 0-length byte array back
binary downloadChunk(1: string id);
// returns json
string getNimbusConf();
// stats functions
ClusterSummary getClusterInfo();
TopologyInfo getTopologyInfo(1: string id) throws (1: NotAliveException e);
//returns json
string getTopologyConf(1: string id) throws (1: NotAliveException e);
StormTopology getTopology(1: string id) throws (1: NotAliveException e);
StormTopology getUserTopology(1: string id) throws (1: NotAliveException e);
}
而對應在Nimbus.java的Java代碼如下,
public class Nimbus { public interface Iface { public void submitTopology(String name, String uploadedJarLocation, String jsonConf, StormTopology topology) throws AlreadyAliveException, InvalidTopologyException, org.apache.thrift7.TException; public void submitTopologyWithOpts(String name, String uploadedJarLocation, String jsonConf, StormTopology topology, SubmitOptions options) throws AlreadyAliveException, InvalidTopologyException, org.apache.thrift7.TException; public void killTopology(String name) throws NotAliveException, org.apache.thrift7.TException; public void killTopologyWithOpts(String name, KillOptions options) throws NotAliveException, org.apache.thrift7.TException; public void activate(String name) throws NotAliveException, org.apache.thrift7.TException; public void deactivate(String name) throws NotAliveException, org.apache.thrift7.TException; public void rebalance(String name, RebalanceOptions options) throws NotAliveException, InvalidTopologyException, org.apache.thrift7.TException; public String beginFileUpload() throws org.apache.thrift7.TException; public void uploadChunk(String location, ByteBuffer chunk) throws org.apache.thrift7.TException; public void finishFileUpload(String location) throws org.apache.thrift7.TException; public String beginFileDownload(String file) throws org.apache.thrift7.TException; public ByteBuffer downloadChunk(String id) throws org.apache.thrift7.TException; public String getNimbusConf() throws org.apache.thrift7.TException; public ClusterSummary getClusterInfo() throws org.apache.thrift7.TException; public TopologyInfo getTopologyInfo(String id) throws NotAliveException, org.apache.thrift7.TException; public String getTopologyConf(String id) throws NotAliveException, org.apache.thrift7.TException; public StormTopology getTopology(String id) throws NotAliveException, org.apache.thrift7.TException; public StormTopology getUserTopology(String id) throws NotAliveException, org.apache.thrift7.TException; }
2 Client
1. 首先Get Client,
NimbusClient client = NimbusClient.getConfiguredClient(conf);
看看backtype.storm.utils下面的client.getConfiguredClient的邏輯,
只是從配置中取出nimbus的host:port, 並new NimbusClient
public static NimbusClient getConfiguredClient(Map conf) { try { String nimbusHost = (String) conf.get(Config.NIMBUS_HOST); int nimbusPort = Utils.getInt(conf.get(Config.NIMBUS_THRIFT_PORT)); return new NimbusClient(conf, nimbusHost, nimbusPort); } catch (TTransportException ex) { throw new RuntimeException(ex); } }
NimbusClient 繼承自ThriftClient, public class NimbusClient extends ThriftClient
ThriftClient又做了什么? 關鍵是怎么進行數據序列化和怎么將數據傳輸到remote
這里看出Thrift對Transport和Protocol的封裝
對於Transport, 其實就是對Socket的封裝, 使用TSocket(host, port)
然后對於protocol, 默認使用TBinaryProtocol, 如果你不指定的話
public ThriftClient(Map storm_conf, String host, int port, Integer timeout) throws TTransportException { try { //locate login configuration Configuration login_conf = AuthUtils.GetConfiguration(storm_conf); //construct a transport plugin ITransportPlugin transportPlugin = AuthUtils.GetTransportPlugin(storm_conf, login_conf); //create a socket with server if(host==null) { throw new IllegalArgumentException("host is not set"); } if(port<=0) { throw new IllegalArgumentException("invalid port: "+port); } TSocket socket = new TSocket(host, port); if(timeout!=null) { socket.setTimeout(timeout); } final TTransport underlyingTransport = socket; //establish client-server transport via plugin _transport = transportPlugin.connect(underlyingTransport, host); } catch (IOException ex) { throw new RuntimeException(ex); } _protocol = null; if (_transport != null) _protocol = new TBinaryProtocol(_transport); }
2. 調用任意RPC
那么就看看submitTopologyWithOpts
client.getClient().submitTopologyWithOpts(name, submittedJar, serConf, topology, opts);
可以看出上面的Nimbus的interface里面有這個方法的定義, 而且Thrift不僅僅自動產生java interface, 而且還提供整個RPC client端的實現
public void submitTopologyWithOpts(String name, String uploadedJarLocation, String jsonConf, StormTopology topology, SubmitOptions options) throws AlreadyAliveException, InvalidTopologyException, org.apache.thrift7.TException { send_submitTopologyWithOpts(name, uploadedJarLocation, jsonConf, topology, options); recv_submitTopologyWithOpts(); }
首先send_submitTopologyWithOpts, 調用sendBase
接着, recv_submitTopologyWithOpts, 調用receiveBase
protected void sendBase(String methodName, TBase args) throws TException { oprot_.writeMessageBegin(new TMessage(methodName, TMessageType.CALL, ++seqid_)); args.write(oprot_); oprot_.writeMessageEnd(); oprot_.getTransport().flush(); } protected void receiveBase(TBase result, String methodName) throws TException { TMessage msg = iprot_.readMessageBegin(); if (msg.type == TMessageType.EXCEPTION) { TApplicationException x = TApplicationException.read(iprot_); iprot_.readMessageEnd(); throw x; } if (msg.seqid != seqid_) { throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, methodName + " failed: out of sequence response"); } result.read(iprot_); iprot_.readMessageEnd(); }
3 Server
Thrift強大的地方是, 實現了整個協議棧而不光只是IDL的轉化, 對於server也給出多種實現
下面看看在nimbus server端, 是用clojure來寫的
可見其中使用Thrift封裝的NonblockingServerSocket, THsHaServer, TBinaryProtocol, Proccessor, 非常簡單
其中processor會使用service-handle來處理recv到的數據, 所以作為使用者只需要在service-handle中實現Nimbus$Iface, 其他和server相關的, Thrift都已經幫你封裝好了, 這里使用的IDL也在backtype.storm.generated, 因為clojure基於JVM所以IDL只需要轉化成Java即可.
(defn launch-server! [conf nimbus]
(validate-distributed-mode! conf)
(let [service-handler (service-handler conf nimbus)
options (-> (TNonblockingServerSocket. (int (conf NIMBUS-THRIFT-PORT)))
(THsHaServer$Args.)
(.workerThreads 64)
(.protocolFactory (TBinaryProtocol$Factory.))
(.processor (Nimbus$Processor. service-handler))
)
server (THsHaServer. options)]
(.addShutdownHook (Runtime/getRuntime) (Thread. (fn [] (.shutdown service-handler) (.stop server))))
(log-message "Starting Nimbus server...")
(.serve server)))