环境：Storm-1.2.2，ubuntu-16.0.4，Idea2018（Linux版）,maven-3.3.9

所有的测试，部署都是在Linux系统上进行。

一、知识点介绍

      个人理解为worker中运行进程，进程中运行线程，线程中运行任务。一个线程可以执行多个任务。并发度就等于所有的任务数（Task）之和。

        现在用代码解释一下：        

TopologyBuilder builder = new TopologyBuilder();
        //设置Spout
        builder.setSpout("wcspout", new WordCountSpout(),3).setNumTasks(4);
        //设置creator-Bolt
        builder.setBolt("split-bolt", new SplitBolt(),4).shuffleGrouping("wcspout").setNumTasks(5);
        //设置counter-Bolt
        builder.setBolt("counter-bolt", new CountBolt(),5).fieldsGrouping("split-bolt", new Fields("word")).setNumTasks(6);
        Config conf = new Config();
        conf.setNumWorkers(2);

        该代码开启了2个worker进程（worker本身不执行Task（任务），它相当于领导，用于产生executor，让executor去执行任务）。给wcspout分配了3个线程4个任务，给split-bolt分配了4个线程5个任务，给counter-bolt分配了5个线程6个任务。

        进程/线程/任务都是平均的。因此对于上面的代码来说，假如开了1个supervisor，那么这2个worker就都由这一个supervisor监管，假如开了2个supervisor，那么每个supervisor管理一个worker。用图形来表示上述代码的任务分配如下：

    每一个task运行一个对象实例。。

       

二、代码

1.CallLogSpout类

该类用于模拟产生数据源

package com.strorm.test;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * @Author zhang
 * @Date 18-6-7 下午2:40
 * Spout类,负责产生数据流
 */
public class CallLogSpout implements IRichSpout {

    //Spout输出收集器
    private SpoutOutputCollector collector;
    //是否完成
    private boolean completed=false;
    //上下文
    private TopologyContext context;
    //随机发生器
    private Random randomGenerator = new Random();

    private Integer idx=0;

    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.context=topologyContext;
        this.collector=spoutOutputCollector;
    }

    public void close() {

    }

    public void activate() {

    }

    public void deactivate() {

    }

    /**
     * 下一个元组
     */
    public void nextTuple() {
        if (idx<=1000){
            List<String> mobileNumbers=new ArrayList<String>();
            mobileNumbers.add("13901645322");
            mobileNumbers.add("13805376831");
            mobileNumbers.add("13500803713");
            mobileNumbers.add("15988321818");
            Integer localIndex=0;
            while (localIndex++<100 && idx<1000){
                //主叫
                String caller=mobileNumbers.get(randomGenerator.nextInt(4));
                //被叫
                String callee=mobileNumbers.get(randomGenerator.nextInt(4));
                while (caller==callee){
                    //主叫与被叫不能相同,重新赋值被叫
                    callee=mobileNumbers.get(randomGenerator.nextInt(4));
                }
                //模拟通话时间
                Integer callTime=randomGenerator.nextInt(60);
                //输出元组
                this.collector.emit(new Values(caller,callee,callTime));

            }
        }

    }

    public void ack(Object o) {

    }

    /**
     *
     * @param o
     */
    public void fail(Object o) {

    }

    /**
     * 定义输出字段
     * @param outputFieldsDeclarer
     */
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //输出元组中到元素为三个，这也要定义三个字段
        outputFieldsDeclarer.declare(new Fields("from","to","callTime"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

2.CallLogBolt类

该类用于处理从Spout类传递过来的源数据，可以实现多个IRichBolt类进行连续处理。

package com.strorm.test;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @Author zhang
 * @Date 18-6-7 下午3:19
 * 创建Bolt
 */
public class CallLogBolt implements IRichBolt {

    private OutputCollector collector;
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector=outputCollector;
    }

    public void execute(Tuple tuple) {
        //处理通话记录
        String from=tuple.getString(0);
        String to=tuple.getString(1);
        Integer callTime=tuple.getInteger(2);
        collector.emit(new Values(from+" 呼叫 "+to,callTime));
    }

    public void cleanup() {

    }

    /**
     * 定义输出字段
     * @param outputFieldsDeclarer
     */
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("call","callTime"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

3.CallCounterBolt类

该类用于统计之前的Bolt类发送过来的数据，其功能类似于Hadoop的Reducer。

package com.strorm.test;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author zhang
 * @Date 18-6-7 下午3:29
 * 计数器,类似于Reducer
 */
public class CallCounterBolt implements IRichBolt {
    Map<String,Integer> counterMap;
    OutputCollector collector;
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.counterMap=new HashMap<String, Integer>();
        collector=outputCollector;
    }

    public void execute(Tuple tuple) {
        String call=tuple.getString(0);
        Integer callTime=tuple.getInteger(1);
        if (!counterMap.containsKey(call)){
            counterMap.put(call,callTime);
        }else {
            Integer integer=counterMap.get(call)+callTime;
            counterMap.put(call,integer);
        }
        collector.ack(tuple);
    }

    public void cleanup() {
        for (Map.Entry<String,Integer> map : counterMap.entrySet()){
            System.out.println(map.getKey()+"   :"+map.getValue()+" 分钟");
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("call"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

4.CallLogOutput类

该类用于提交Topology，类似于Hadoop的Job提交。由于Storm不会停止数据流作业，所以在测试环境下要人为停止。这里的Sleep时间可以根据实际情况设定。如果时间过短，可能看不到输出结果。

package com.strorm.test;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @Author zhang
 * @Date 18-6-7 下午3:46
 */
public class CallLogOutput {

    public static void main(String[] args) throws InterruptedException, InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        TopologyBuilder topologyBuilder=new TopologyBuilder();
        //设置Spout
        topologyBuilder.setSpout("spout",new CallLogSpout());
        //设置Bolt
        topologyBuilder.setBolt("bolt",new CallLogBolt()).shuffleGrouping("spout");
        //设置counterBolt
        topologyBuilder.setBolt("counterBolt",new CallCounterBolt()).fieldsGrouping("bolt",new Fields("call"));
        Config config=new Config();
        config.setDebug(true);
        LocalCluster localCluster=new LocalCluster();
        localCluster.submitTopology("Log",config,topologyBuilder.createTopology());
        Thread.sleep(20000);
        localCluster.shutdown();
    }
}

5.输出结果

根据代码中的随机产生4个号码。再根据排列组合的

可以知道输出是正确的。

6.集群部署执行

集群部署执行需要将localCluster提交改成StormSubmitter.submitTopology提交。并将相关的Module打包成jar包。

以本文的代码为例，部署时，执行：

storm jar storm-call-core-1.0-SNAPSHOT.jar com.strorm.test.CallLogOutput

在上图中，可以看到提交的Topology名字为Log，就是代码中设定的Topology名字。

拓扑图：

可以看到上图中三个红色圈中拓扑节点的名字就是代码中设定的名字。

然后回到web ui的主页可以看到Topology已经没有数据信息了，表示没有Topology在执行。

标准的执行格式为：

storm jar jar包名  xxx.类名 [arg1] [arg2] [arg3]...

当在集群上执行了jar包后，会在某个supervisor的目录下，产生一个worker.log，这个日志文件里面记录了输出结果，但是该文件一般会非常的大。

源代码下载：点击下载storm电话日志分析源代码

下载源码完成后，只需要以导入maven工程的方式导入root目录下的 pom.xml文件，然后会自动引入所有相关的module。