程序员社区

Storm中几种基本的Bolt接口的特点

Storm中几种基本的Bolt接口的特点

Storm中定义的Bolt接口主要有IBolt 、 IRichBolt 、 IBasicBolt和IBatchBolt,先看一下类图

Storm中几种基本的Bolt接口的特点插图
bolt类图

一.IRichBolt

Storm中最常用来定义Topology组件的接口。 它十分灵活, 用户可以通过其实现各种控制逻辑, 并且能控制何时进行Ack 、 Fail和Anchor操作。

Bolt是Storm中的基础运行单位, 当其启动并有消息输人时, 将调用execute方法来进行处理。与ISpout类似, IBolt对象在提交时也会被序列化为字节数组, 具体的执行节点通过反序列化的方法得到该对象, 并调用prepare回调方法。用户应将复杂对象的初始化放在prepare回调方法中实现, 以保证每个具体对象都可以正确初始化。对象被销毁时, 将调用cleanup回调方法, 但是Storm并不保证该方法一定被执行。

通常, 在execute方法的实现中会对输人消息进行处理, 这有可能产生新消息需要发送到下游节点, 最后还要对输入的消息进行Ack操作。 如果消息处理失败, 则需对输入的消息进行Fail操作, 这是保证Ack消息系统可以正常工作的基础。

二.IBasicBolt

Storm中提供的定义简单逻辑的Topology组件接口,用户基于它实现自己的Bolt也比较简单。

基于IBasicBolt编写的好处是Storm框架本身帮你处理了所发出消息的Ack 、 Fail和Anchor操作, 这是由执行器BasicBoltExecutor实现的。BasicBoltExecutor实现了IRichBolt接口, 同时还包含了一个IBasciBolt成员变量用于调用的转发。 它是基于装饰模式实现的。

其代码如下所示

public class BasicBoltExecutor implements IRichBolt {
    public static Logger LOG = LoggerFactory.getLogger(BasicBoltExecutor.class);
    private IBasicBolt _bolt;
    private transient BasicOutputCollector _collector;

    public BasicBoltExecutor(IBasicBolt bolt) {
        //内部封装了真实的bolt
        this._bolt = bolt;
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        this._bolt.declareOutputFields(declarer);
    }

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this._bolt.prepare(stormConf, context);
        this._collector = new BasicOutputCollector(collector);
    }

    public void execute(Tuple input) {
        //该execute方法都是通过调用真实的bolt来完成的
        this._collector.setContext(input);

        try {
            this._bolt.execute(input, this._collector);
            this._collector.getOutputter().ack(input);
        } catch (FailedException var3) {
            if (var3 instanceof ReportedFailedException) {
                this._collector.reportError(var3);
            }

            this._collector.getOutputter().fail(input);
        }

    }

    public void cleanup() {
        this._bolt.cleanup();
    }

    public Map<String, Object> getComponentConfiguration() {
        return this._bolt.getComponentConfiguration();
    }
}

但IBasicBolt的使用是有限制的, 基于收到的某条消息衍生出来的所有消息必须在一次execute中发送出去, 否则内置的Ack机制将不能保证Bolt的正常工作。 所以, 用户应该避免使用该类型的Bolt来做诸如聚集或者连接的操作。

三.IBatchBolt

它是Storm提供的用来处理批量数据的接口。 目前,它只用于事务Topology中, 它是Storm实现事务Topology的基础。 IBatchBolt在系统收到属于某Batch的第一条消息时被创建, 而在所有的消息都处理完成之后再被销毁。 Storm中采用反序列化对象的方式来弥补不断创建IBatchBolt对象所带来的负担。

具有finishBatch方法: 该方法仅当这批消息被处理完时才会被调用。 如果BatchBolt同时实现了ICoranitter的接口, finishBatch方法只有当该Batch之前的所有Batch均被成功处理后才被调用。 这保证了强序关系,同时也是Storm中事务Topology的实现基础。

赞(0) 打赏
未经允许不得转载:IDEA激活码 » Storm中几种基本的Bolt接口的特点

相关推荐

  • 暂无文章

一个分享Java & Python知识的社区