初识Jstorm 多个bolt应用

最近接到任务说要使用jstorm处理业务，之前没接触过，只能硬着头皮来，接下来谈谈我这两天的收获

 

1，怎么了解jstorm，这个答案没什么固定的，但是我个人比较喜欢去看官方的文档，如果官方的文档实在找不到了，再去baidu，baidu不到那就只能google了，如果还是找不到。。没办法，看源代码吧；

 

      对于官方文档，我现在能找到的只有：

 

      https://github.com/alibaba/jstorm/wiki/JStorm-Chinese-Documentation 

 

      该地址是jstorm的中文文档，上面对jstorm做了简要的概述，内容并不是很多，看完了， 至少能对jstorm有个大体的了解

 

      因为我看博客的时候，喜欢在博客上直接找到答案，所以我在这里赘述一下

 

     根据ali的文档得知，jStorm是一个分布式实时计算引擎，jstorm是 从storm发展而来，由ali爸爸开源的，是一个类似Hadoop MapReduce的系统，用户创建任务后，提交给jstorm系统，JStorm系统会将这个任务跑起来，7*24不间断，如果一个worker发生一 碗，立即分配一个worker来替换失效的worker（这段文字摘自 alibaba的github），从系统的角度来说，jstorm是类似Mapreduce的调度系统，从数据的角度来说，Jstorm是一套基于流水线 的消息处理机制；

 

说到流水线，你能想到什么？至少从流水两个字，我想到了管道，其实个人理解jstorm的流水线很像是管道，想象一下，一个主管道可以分出多个支流管道，也可以接一个弯管或者其他什么不一样的管，来看张图

 

这个图完全是偷ali的，在看另一张

 

 

也是偷的，从这两张图就可以形象的看出jstorm的思想了

 

接下来解释几个重要的概念：

 

1，topology

 

    我的英文差，只能音译了。。拓扑结构图，其实也差不多是这个意思

 

    jstorm所执行的任务，其实就是一个topology，一个topology可以包含多个spout，多个bolt，多级bolt

 

2，spout

 

   流的来源

 

3，bolt

 

    可以说是流的去处（这里可以进行业务处理）

 

 

 

简单来说：在jstorm运行环境正常的情况下，开发人员要做的就是

 

     建立一个topology，建立若干spout，建立若干bolt，然后将spout加入到topology，将bolt也加入到topology，这时组成了一个相对完整的拓扑结构图，然后提交给jtorm，齐活！

 

说起来比较简单，那么实际写起来呢，来看下面的代码

 

注意⚠️：有个前提是jstorm运行环境正常，

 

首先是配置文件：看过其他博客的人知道有*.ynml的文件，主要是内容是对拓扑结构的配置，这个建议新手可以直接忽略，先不做修改，仔细研读官方的文档之后再尝试

 

经过上面的解释，我们知道一个拓扑结构包涵哪些东西了。其实代码也是差不多的

 

首先：一个TestSpout.java

 

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class TestSpout extends BaseRichSpout {
    
    private static final long serialVersionUID = 4924189748702648696L;

    private static final Logger LOG = LoggerFactory.getLogger(TestSpout.class);
    
    SpoutOutputCollector collector;
    
    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;

    }

    @Override
    public void nextTuple() {
        this.collector.emit(new Values("test123"));
//        this.collector.emit("VEHICLEINFO", new Values("test123"));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // TODO Auto-generated method stub
        declarer.declare(new Fields("VEHICLE"));
//        declarer.declareStream("VEHICLEINFO", new Fields("VEHICLE"));
    }

}

 

继承自BaseRichSpout，需要覆盖三个方法，分别来说下：

 

open：这个应该是在创建拓扑结构，提交的时候会调用一次，具体我没细研究，基本都是这么写

 

nextTuple：发送数据的源头，发送的方式是调用SpoutOutputCollector.emit方法，数据的方式就是用Values做 封装，这里注意下：jstorm的流支持任何可序列化的数据类型，包括bean，只要是可序列化的就可以，数据的来源，看你的应用场景，可以来自数据库， 也可以来自mq，或者是Kafka等等，我们做测试，直接写死就好了

 

declareOutputFields： 这里一开始一直没搞明白是干嘛的，通过无数次失败，发现，其实他是emit对应的，采用未注视的方式提交的时候，在此方法中 new Fields的值启示就是对应Values的值，是一个建值对，在bolt里面可以通过input.getValueByFiled()的方式获取 （get方法名我写了个大概，可能写错，各位亲自己 点 一下就出来了）

 

那么注视的掉的是什么情况呢，首先注视掉的emit第一个参数是streamId，就相当于我指定要发给谁，本方法中注视掉的是和emit对应的

 

 

 

2，来一个TestBolt

 

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class TestBolt extends BaseRichBolt {

    private static final long serialVersionUID = 1804939402092249890L;

    private static final Logger LOG = LoggerFactory.getLogger(TestBolt.class);
    
    OutputCollector collector;
    
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String s = input.getString(0);
        String sf = input.getStringByField("VEHICLE");
        LOG.error("receive MSG----------------------" + s + " ---- " + sf);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }

}

 

这个只有方法execute可以说一说，就是获取数据，tuple就是一个数据流，里面有你的数据，你可以通过get参数序号的方式，也可以用Field方式，而这个getValueByField的值就是上面spout中定义的，有兴趣的自己回顾一下

 

3，来一个TestTopology

 

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

import com.zuche.jstom.example.topology.spout.TestSpout;
import com.zuche.jstom.example.topology.spout.bolt.TestBolt;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

public class TestTopology {
    public static void main(String[] args) {
        
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("testspout", new TestSpout(), 1);
        builder.setBolt("testbolt", new TestBolt(), 1).shuffleGrouping("testspout");
//        builder.setBolt("testbolt", new TestBolt(), 1).shuffleGrouping("testspout", "VEHICLEINFO");
        
        Map<Object,Object> conf = new HashMap<Object, Object>();
        
        conf.put(Config.TOPOLOGY_WORKERS, 3);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("testTopology", conf, builder.createTopology());
    }
}

 

 

 

这个就是拼装topology的过程，仔细看的发现有个注视的，这个注视的就是上面说的，我指定这个bolt处理哪个streamId的数据，和前面的spout对应起来，仔细看看

 

接下里就是直接跑就行了

 

这里说下：如果本地搭建了环境，跑的时候发现有个slf4j的版本不支持的错误信息，是因为jstorm使用了1.4的版本slf4j，你把她们exclude掉，然后自己引入一个高版本的就行了

 

 

 

那么来说说多个bolt的情况

 

其实就是在上面的代码再加一行代码：

 

builder.setBolt("testbolt2", new TestBolt2(), 1).shuffleGrouping("testbolt");

 

记得自己写个TestBolt2，这是数据流就是 testspout > testbolt > testbolt2

 

这里要说下shuffleGrouping，jstorm中有几个类似后缀的方法，具体的含义大家看文档，我就不多说了，他们的功能并不相同

 

可以说为了实现同一个目的，可以有好几种写法，但是具体哪个好，还是大家自己体会，我也是新手，不懂。。

 

举个例子：

 

        builder.setSpout(LocalSpout.LOCAL_SPOUT_NAME, new LocalSpout(), 1);
        
        builder.setBolt(FenceBolt.FENCE_BOLT_NAME, new FenceBolt(), 1).shuffleGrouping(LocalSpout.LOCAL_SPOUT_NAME);
        
//        builder.setBolt(AlarmBolt.ALARM_BOLT_NAME, new AlarmBolt(), 1).shuffleGrouping(FenceBolt.FENCE_BOLT_NAME);
//        builder.setBolt(AlarmBolt.ALARM_BOLT_NAME, new AlarmBolt(), 1).fieldsGrouping(FenceBolt.FENCE_BOLT_NAME , new Fields("FENCEVEHICLE"));
        builder.setBolt(AlarmBolt.ALARM_BOLT_NAME, new AlarmBolt(), 1).shuffleGrouping(FenceBolt.FENCE_BOLT_NAME, FenceBolt.FENCE_STREAM_ID);

一个spout，两个连续的bolt，第二个bolt的配置方式，三种写法都可以

 

深层次的区别我还没体会出来，等我体会出来了，再和大家交流

到此，本文结束，有时间话，我会再写一篇介绍概念的