JStorm与Storm源码分析（五）--SpoutOutputCollector与代理模式

/**
 * ISpoutOutputCollector:Spout输出收集器接口
 */
public interface ISpoutOutputCollector {
    /**
     * 改方法用来向外发送数据,它的返回值是该消息所有发送目标的taskID集合;
     * 参数:
     * streamId:消息Tuple将要被输出到的流
     * tuple:要输出的消息,是一个Object列表
     * messageId:输出消息的标记信息,如果messageId被设置为null,则Storm不会追踪该消息,
     * 否则它会被用来追踪所发出的消息处理情况
     */
    List emit(String streamId, List

public class SpoutOutputCollector implements ISpoutOutputCollector {
    /**
     * 持有SpoutOutputCollector要代理的对象
     */
    ISpoutOutputCollector _delegate;

    public SpoutOutputCollector(ISpoutOutputCollector delegate) {
        _delegate = delegate;
    }
    /**
     * 实现了接口中的emit方法,并且提供了它的几个重载方法
     * eg.如果不指定streamId,默认使用default,如果不指定messageId,则默认使用空(null)
     */
    public List emit(String streamId, List

//抽象主题
public interface Subject {
    public void process(String taskName);
}

public class RealSubject implements Subject {
    @Override
    public void process(String taskName) {
        System.out.println("正在执行任务:"+taskName);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

public class ProxySubject implements Subject {
    //代理类持有一个委托类的对象引用
    private Subject delegate;
    public ProxySubject(Subject delegate){
        this.delegate=delegate;
    }
    @Override
    public void process(String taskName) {
        //预处理
        this.before();
        //将请求分派给委托类处理
        delegate.process(taskName);
        //善后处理
        this.after();
    }
    private void before(){
        System.out.println("预处理!");
    }
    private void after(){
        System.out.println("善后处理!");
    }
}

public class Test {
    public static void main(String[] args) {
        RealSubject subject = new RealSubject();
        ProxySubject p = new ProxySubject(subject);
        p.process("排水");
    }
}

预处理!
正在执行任务:排水
善后处理!

public interface Service {
    //目标方法 
    public void process();
}

public class UserServiceImpl implements Service {
    @Override
    public void process() {
         System.out.println("用户service处理");  
    }
}

public class MyInvocatioHandler implements InvocationHandler {
    private Object target;

    public MyInvocatioHandler(Object target) {
        this.target = target;
    }
    @Override
    public Object invoke(Object proxy, Method method, Object[] args)
            throws Throwable {
        //System.out.println("-----before-----");
        this.before();
        Object result = method.invoke(target, args);
       // System.out.println("-----end-----");
        this.after();
        return result;
    }
    // 生成代理对象
    public Object getProxy() {
        ClassLoader loader = Thread.currentThread().getContextClassLoader();
        Class>[] interfaces = target.getClass().getInterfaces();
        return Proxy.newProxyInstance(loader, interfaces, this);
    }

    private void before(){
        System.out.println("预处理!");
    }
    private void after(){
        System.out.println("善后处理!");
    }
}

public class ProxyTest {
    public static void main(String[] args) {
        Service service = new UserServiceImpl();
        MyInvocatioHandler handler = new MyInvocatioHandler(service);
        Service serviceProxy = (Service)handler.getProxy();
        serviceProxy.process();
    }
}

预处理!
用户service处理
善后处理!