import backtype.storm.generated.Nimbus.Client; //导入依赖的package包/类
protected long getTopologyTPS(TopologySummary topology, Client client) throws NotAliveException, TException{
    long topologyTps = 0l;
    String topologyId = topology.get_id();
    if(topologyId.startsWith("ClusterMonitor")){
        return topologyTps;
    }
    TopologyInfo topologyInfo = client.getTopologyInfo(topologyId);
    if(topologyInfo == null){
        return topologyTps;
    }
    List<ExecutorSummary> executorSummaryList = topologyInfo.get_executors();
    for(ExecutorSummary executor : executorSummaryList){
        topologyTps += getComponentTPS(executor);
    }
    LOGGER.info("topology = " + topology.get_name() + ", tps = " + topologyTps);
    return topologyTps;
}
 

import backtype.storm.generated.Nimbus.Client; //导入依赖的package包/类
public static List nimbusClientandConn(String host, Integer port)
		throws TTransportException {
	TSocket ts = new TSocket(host, port);
	TFramedTransport tt = new TFramedTransport(ts);
	TBinaryProtocol prot = new TBinaryProtocol(tt);
	Client nc = new Client(prot);
	ts.open();
	List l = new ArrayList();
	l.add(nc);
	l.add(tt);
	return l;
}
 

import backtype.storm.generated.Nimbus.Client; //导入依赖的package包/类
private void getClusterInfo(Client client) {
    try {
        ClusterSummary clusterSummary = client.getClusterInfo();
        List<SupervisorSummary> supervisorSummaryList = clusterSummary.get_supervisors();
        int totalWorkers = 0;
        int usedWorkers = 0;
        for(SupervisorSummary summary : supervisorSummaryList){
            totalWorkers += summary.get_num_workers() ;
            usedWorkers += summary.get_num_used_workers();
        }
        int freeWorkers = totalWorkers - usedWorkers;
        LOGGER.info("cluster totalWorkers = " + totalWorkers 
                + ", usedWorkers = " + usedWorkers 
                + ", freeWorkers  = " +  freeWorkers);
        
        HttpCatClient.sendMetric("ClusterMonitor", "freeSlots", "avg", String.valueOf(freeWorkers));
        HttpCatClient.sendMetric("ClusterMonitor", "totalSlots", "avg", String.valueOf(totalWorkers));
        
        List<TopologySummary> topologySummaryList = clusterSummary.get_topologies();
        long clusterTPS = 0l;
        for(TopologySummary topology : topologySummaryList){
            long topologyTPS = getTopologyTPS(topology, client);
            clusterTPS += topologyTPS;
            if(topology.get_name().startsWith("ClusterMonitor")){
                continue;
            }
            HttpCatClient.sendMetric(topology.get_name(), topology.get_name() + "-TPS", "avg", String.valueOf(topologyTPS));
        }
        HttpCatClient.sendMetric("ClusterMonitor", "ClusterEmitTPS", "avg", String.valueOf(clusterTPS));
        
    } catch (TException e) {
        initClient(configMap);
        LOGGER.error("get client info error.", e);
    }
    catch(NotAliveException nae){
        LOGGER.warn("topology is dead.", nae);
    }
}
 

import backtype.storm.generated.Nimbus.Client; //导入依赖的package包/类
/**
 * Creates a connection with explicit nimbus host. Ignored if a local cluster is set 
 * for testing.
 * 
 * @param nimbusHost the nimbus host name
 * @param port the nimbus port number
 * @see #setLocalCluster(ILocalCluster)
 */
public ThriftConnection(String nimbusHost, int port) {
    if (null == localCluster) {
        socket = new TSocket(nimbusHost, port);
        LOGGER.info("Thrift connection info " + nimbusHost + " " + port);
        transport = new TFramedTransport(socket);
        protocol = new TBinaryProtocol(transport);
        client = new Client(protocol);
    }
}