聊聊puma的Dispatcher

序

本文主要研究一下puma的Dispatcher

Dispatcher

puma/puma/src/main/java/com/dianping/puma/sender/dispatcher/Dispatcher.java

public interface Dispatcher extends LifeCycle {
    String getName();

    void dispatch(ChangedEvent event, PumaContext context) throws DispatcherException;

    List<Sender> getSenders();
}
複製代碼

• Dispatcher定義了getName、dispatch、getSenders方法

AbstractDispatcher

puma/puma/src/main/java/com/dianping/puma/sender/dispatcher/AbstractDispatcher.java

public abstract class AbstractDispatcher implements Dispatcher {
    private String name;

    /*
     * (non-Javadoc)
     *
     * @see com.dianping.puma.common.LifeCycle#start()
     */
    @Override
    public void start() {

    }

    /*
     * (non-Javadoc)
     *
     * @see com.dianping.puma.common.LifeCycle#stop()
     */
    @Override
    public void stop() {

    }

    /*
     * (non-Javadoc)
     *
     * @see com.dianping.puma.sender.dispatcher.Dispatcher#getName()
     */
    @Override
    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    protected void throwExceptionIfNeeded(List<Throwable> exceptionList) throws DispatcherException {

        if (exceptionList != null && !exceptionList.isEmpty()) {
            StringWriter buffer = new StringWriter();
            PrintWriter out = null;
            try {
                out = new PrintWriter(buffer);

                for (Throwable exception : exceptionList) {
                    exception.printStackTrace(out);
                }
            } finally {
                if (out != null) {
                    out.close();
                }
            }

            throw new DispatcherException(buffer.toString());
        }

    }

}
複製代碼

• AbstractDispatcher定義了throwExceptionIfNeeded方法，它將exceptionList轉換爲DispatcherException

SimpleDispatcherImpl

puma/puma/src/main/java/com/dianping/puma/sender/dispatcher/SimpleDispatcherImpl.java

public class SimpleDispatcherImpl extends AbstractDispatcher {

    private static final Logger log = Logger.getLogger(SimpleDispatcherImpl.class);

    private List<Sender> senders;

    /**
     * @return the senders
     */
    public List<Sender> getSenders() {
        return senders;
    }

    /**
     * @param senders
     *           the senders to set
     */
    public void setSenders(List<Sender> senders) {
        this.senders = senders;
    }

    @Override
    public void start() {
        for (Sender sender : senders) {
            sender.start();
        }
        super.start();
    }

    @Override
    public void stop() {
        for (Sender sender : senders) {
            sender.stop();
        }
        super.stop();
    }

    @Override
    public void dispatch(ChangedEvent event, PumaContext context) throws DispatcherException {
        if (senders != null && senders.size() > 0) {
            List<Throwable> exceptionList = new ArrayList<Throwable>();
            for (Sender sender : senders) {
                try {
                    sender.send(event, context);
                } catch (Exception e) {
                    log.error("Exception occurs in sender " + sender.getName());
                    exceptionList.add(e);
                }
            }

            throwExceptionIfNeeded(exceptionList);
        } else {
            log.warn("No senders in dispatcher " + getName());
        }
    }

}
複製代碼

• SimpleDispatcherImpl繼承了AbstractDispatcher，其start方法遍歷senders，挨個執行sender.start()；其stop方法遍歷senders，挨個執行sender.stop()；其dispatch方法遍歷senders，挨個執行sender.send(event, context)

小結

Dispatcher定義了getName、dispatch、getSenders方法；AbstractDispatcher定義了throwExceptionIfNeeded方法，它將exceptionList轉換爲DispatcherException；SimpleDispatcherImpl繼承了AbstractDispatcher，實現了start、stop、dispatch方法

doc

• Dispatcher