JStorm与Storm源码分析（七）--BasicBoltExecutor与装饰模式

/**
 * BasicBoltExecutor实现了IRichBolt接口
 * 在该类中持有一个IBasicBolt成员变量用于调用转发
 * 说明:
 * 该类是基于装饰模式实现的.
 */
public class BasicBoltExecutor implements IRichBolt {
    public static Logger LOG = LoggerFactory.getLogger(BasicBoltExecutor.class);    
    //持有IBasicBolt类型的变量
    private IBasicBolt _bolt;
    //定义了成员变量_collector
    private transient BasicOutputCollector _collector;

    public BasicBoltExecutor(IBasicBolt bolt) {
        _bolt = bolt;
    }
    /**
     * 实现declareOutputFields方法,但它实际上是
     * _bolt调用declareOutputFields方法
     */
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        _bolt.declareOutputFields(declarer);
    }

    /**
     * 实现prepare方法,
     * 实际上是调用_bolt的prepare方法,
     * 并实例化BasicOutputCollector
     */
    public void prepare(Map stormConf, TopologyContext context, 
            OutputCollector collector) {
        _bolt.prepare(stormConf, context);
        _collector = new BasicOutputCollector(collector);
    }
    /**
     * 实现execute方法
     */
    public void execute(Tuple input) {
        //设置运行器上下文,
        //它表示经execute方法发送出去的消息都是由输入消息产生的,
        //即输出的消息都将标记为输入消息所衍生出来的消息,
        //这是使用IBasicBolt实现消息跟踪的重要一环
        _collector.setContext(input);
        try {
            //调用_bolt的execute方法
            _bolt.execute(input, _collector);
            //对输入的消息进行Ack操作.
            //这一步意味着基于当前输入消息的处理和衍生消息的发送已经完成,
            //这时就可以对该消息进行Ack操作了.
            _collector.getOutputter().ack(input);
        } catch(FailedException e) {
            //Storm捕获所有的FailedException,并对输入的消息进行Fail操作。
            //如果捕获的异常为ReportedFailedException的实例,
            //则调用reportError回调方法,给用户一个机会去处理异常
            //FailedException是Storm定义的一种基本异常,用来进行消息的失败重发等操作,
            //并不会导致Topology运行停止
            if(e instanceof ReportedFailedException) {
                _collector.reportError(e);
            }
            _collector.getOutputter().fail(input);
        }
    }
    public void cleanup() {
        _bolt.cleanup();
    }
    public Map getComponentConfiguration() {
        return _bolt.getComponentConfiguration();
    }
}

/**
 * Component  
 * 抽象构建角色,
 * 约束或规范准备接收附加责任的对象
 */
public interface Component {
    public void operation();
}

/**
 * ConcreteComponent  
 * 接收附加责任
 */
public class ConcreteComponent implements Component {

    @Override
    public void operation() {
        System.out.println("的咖啡");
    }
}

/**
 * Decorator装饰角色 
 */
public abstract class Decorator implements Component {
    //持有一个构建对象的实例
    private Component component;
    public Decorator(Component component){
        this.component=component;
    }
    @Override
    public void operation() {
        component.operation();
    }
}

/** 
 * ConcreteDecorator1
 * 加糖装饰器
 */
public class ConcreteDecorator1 extends Decorator {

    public ConcreteDecorator1(Component component) {
        super(component);
        // TODO Auto-generated constructor stub
    }
    public void operation(){
        System.out.print("加糖");
        super.operation();
    }
}

/**
 * ConcreteDecorator2   
 * 加奶装饰器
 */
public class ConcreteDecorator2 extends Decorator {

    public ConcreteDecorator2(Component component) {
        super(component);
        // TODO Auto-generated constructor stub
    }
    public void operation(){
        System.out.print("加奶");
        super.operation();
    }
}

public class Test {
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        Component component=new ConcreteComponent();
        Decorator decorator1=new ConcreteDecorator2(component);
        Decorator decorator2=new ConcreteDecorator1(decorator1);
        decorator2.operation();
    }
}

加糖加奶的咖啡