责任链模式

写在前面的话

很多的系统在不断迭代的过程中，免不了代码的默认规范越来越不可控，越来越堆积，这里希望近一点绵薄之力，希望既能相对规范下代码的逻辑，又可以方便后来者更好的融入前人的规范中，并轻松的进行扩展和优化。

注：
相关系列文章会不定期更新（主要因为我比较懒），有想法的读者可以留言探讨哈~

另外，如有引用或者拷贝的请注明出处。

废话不多说了，下面进入新的废话中~

责任链模式

思考部分

一旦说道责任链模式，我们能想到的需要的组件有handler（处理器）、context（上下文，用于控制是否继续下一步等）、pipeline（管道，用于连接多个处理器）。
既然责任链的组件基本固定，那我们是不是可以写一套通用责任链模板，这样在需要使用责任链模式的地方直接实现自己的处理类就可以，从而避免每次都重新写一套模板。

好，接下来我们定义这一套通用责任链模板为channel（借鉴了Netty里的责任链概念）。

上代码

泛型I代表入参，O代表出参。

1、先上一套接口，定义各个组件

IChannelHandler<I, O>事件处理器

    /**
     * 具体处理内容
     */
    void process(IChannelHandlerContext<I, O> ctx, I in, O out) throws Exception;

    /**
     * 发生异常时触发
     */
    void exceptionCaught(IChannelHandlerContext<I, O> ctx, Throwable cause, I in, O out) throws Exception;

IChannelHandlerContext<I, O>上下文定义

    /**
     * 当前处理器
     */
    IChannelHandler<I, O> handler();

    /**
     * 继续下一个处理器处理，类似doFilter
     */
    IChannelHandlerContext<I, O> fireChannelProcess(I in, O out);

    /**
     * 把异常交给下一个处理器处理
     */
    IChannelHandlerContext<I, O> fireExceptionCaught(Throwable cause, I in, O out);

IChannelPipeline<I, O>链定义

    /**
     * 添加处理器
     */
    IChannelPipeline<I, O> append(IChannelHandler<I, O> handler);

    /**
     * 启动事件处理
     */
    IChannelPipeline<I, O> run(I in, O out);

IChannel<I, O>通道（责任链）定义

    /**
     * 处理器管道
     */
    IChannelPipeline<I, O> pipeline();

    /**
     * 处理入口，这里直接调用pipeline的run方法
     */
    IChannel<I, O> process(I in, O out);

到这里，一套责任链模式的模板接口基本定义完成。既然要实现一套通用的责任链模板，那就需要实现一套默认的实现类，然后不同业务逻辑去实现自己的处理器。

2、默认的模板实现

IChannelHandler<I, O>事件处理器

这里默认适配了一个事件处理器，具体业务处理时继承此类或者实现接口都可以。

public class ChannelHandlerAdapter<I, O> implements IChannelHandler<I, O> {

    @Override
    public void process(IChannelHandlerContext<I, O> ctx, I in, O out) throws Exception {
        ctx.fireChannelProcess(in, out);
    }

    @Override
    public void exceptionCaught(IChannelHandlerContext<I, O> ctx, Throwable cause, I in, O out) throws Exception {
        ctx.fireExceptionCaught(cause, in, out);
    }
}

IChannelHandlerContext<I, O>上下文定义

实现上下文之前，先定义一个执行处理器的工具类。供上下文执行当前处理器或调起下一个处理器时使用。

public class ChannelHandlerUtil {

    public static void invokeChannelProcess(IChannelHandlerContext ctx, Object in, Object out) {
        try {
            ctx.handler().process(ctx, in, out);
        } catch (Throwable cause) {
            invokeExceptionCaught(ctx, cause, in, out);
        }
    }

    public static void invokeExceptionCaught(IChannelHandlerContext ctx, Throwable cause, Object in, Object out) {
        try {
            ctx.handler().exceptionCaught(ctx, cause, in, out);
        } catch (Exception ex) {
            log.error("[invokeExceptionCaught] failed. ctx={}, in={}, out={}, cause={}", ctx, in, out, cause, ex);
        }
    }
}

抽象的上下文模板

@Data
public abstract class AbstractChannelHandlerContext<I, O> implements IChannelHandlerContext<I, O> {

    AbstractChannelHandlerContext<I, O> prev;
    AbstractChannelHandlerContext<I, O> next;

    @Override
    public IChannelHandlerContext<I, O> fireChannelProcess(I in, O out) {
        ChannelHandlerUtil.invokeChannelProcess(next, in, out);
        return this;
    }

    @Override
    public IChannelHandlerContext<I, O> fireExceptionCaught(Throwable cause, I in, O out) {
        ChannelHandlerUtil.invokeExceptionCaught(next, cause, in, out);
        return this;
    }

}

上下文的默认实现

public class DefaultChannelHandlerContext<I, O> extends AbstractChannelHandlerContext<I, O> {

    private final IChannelHandler<I, O> handler;

    public DefaultChannelHandlerContext(IChannelHandler<I, O> handler) {
        this.handler = handler;
    }

    @Override
    public IChannelHandler<I, O> handler() {
        return handler;
    }
}

IChannelPipeline<I, O>链定义

默认的处理器链。默认提供首位节点。

public class DefaultChannelPipeline<I, O> implements IChannelPipeline<I, O> {

    AbstractChannelHandlerContext<I, O> head;
    AbstractChannelHandlerContext<I, O> tail;

    public DefaultChannelPipeline() {
        head = new HeadContext<>();
        tail = new TailContext<>();

        head.setNext(tail);
        tail.setPrev(head);
    }

    @Override
    public IChannelPipeline<I, O> append(IChannelHandler<I, O> handler) {
        AbstractChannelHandlerContext<I, O> newCtx = new DefaultChannelHandlerContext(handler);
        AbstractChannelHandlerContext<I, O> prev = tail.getPrev();

        newCtx.setPrev(prev);
        newCtx.setNext(tail);
        tail.setPrev(newCtx);
        prev.setNext(newCtx);
        return this;
    }

    @Override
    public IChannelPipeline<I, O> run(I in, O out) {
        ChannelHandlerUtil.invokeChannelProcess(head, in, out);
        return this;
    }

    @Slf4j
    final static class HeadContext<I, O> extends AbstractChannelHandlerContext<I, O> implements IChannelHandler<I, O> {

        @Override
        public void process(IChannelHandlerContext<I, O> ctx, I in, O out) throws Exception {
            log.info("[bootstrap] start. in={}, out={}", in, out);
            ctx.fireChannelProcess(in, out);
        }

        @Override
        public void exceptionCaught(IChannelHandlerContext<I, O> ctx, Throwable cause, I in, O out) throws Exception {
            ctx.fireExceptionCaught(cause, in, out);
        }

        @Override
        public IChannelHandler<I, O> handler() {
            return this;
        }
    }

    @Slf4j
    final static class TailContext<I, O> extends AbstractChannelHandlerContext<I, O> implements IChannelHandler<I, O> {

        @Override
        public void process(IChannelHandlerContext<I, O> ctx, I in, O out) throws Exception {
            log.info("[bootstrap] end. in={}, out={}", in, out);
        }

        @Override
        public void exceptionCaught(IChannelHandlerContext<I, O> ctx, Throwable cause, I in, O out) throws Exception {
            log.error("[bootstrap] end failed. in={}, out={}", in, out, cause);
        }

        @Override
        public IChannelHandler<I, O> handler() {
            return this;
        }
    }
}

IChannel<I, O>通道（责任链）定义

public class DefaultChannel<I, O> implements IChannel<I, O> {

    private IChannelPipeline<I, O> pipeline;

    public DefaultChannel() {
        pipeline = new DefaultChannelPipeline<>();
    }

    @Override
    public IChannelPipeline<I, O> pipeline() {
        return pipeline;
    }

    @Override
    public IChannel<I, O> process(I in, O out) {
        pipeline.run(in, out);
        return this;
    }
}

好，到这里基于通用模板的实现已经完成。

3、简单使用

这里需要实现具体的处理器（IChannelHandler）

// 定义事件对应的链（只需要执行一次）
IChannel<String, String> channel = new DefaultChannel();
// 加入自定义的处理器（只需要执行一次）
channel.pipeline().append(handler...);

// 执行处理
channel.process("hello", "world");

这里假设的入参出参都是string。

4、升级版

如果把每一套入参出参定义为一类事件，那么在每一类事件只需要初始化一次IChannel，之后每次事件只需要调用process方法的背景下，是不是可以进一步简化模板使用？是的，实现一套调度器，把所有事件处理器都注册到调度器中，需要执行事件时，直接调用调度器即可。

IChannelDispatcher调度器定义

    /**
     * 注册事件处理器
     */
    <I, O> IChannelDispatcher register(IChannelHandler<I, O>... handlers);

    /**
     * 开始处理对应事件
     */
    <I, O> void process(I in, O out);

IChannelDispatcher调度器默认实现

@Slf4j
public class DefaultChannelDispatcher implements IChannelDispatcher {


    private Map<Key, IChannel> channelMap = new HashMap<>();


    @Override
    public <I, O> IChannelDispatcher register(IChannelHandler<I, O>... handlers) {
        List<Class> genericitys = GenericityUtil.getGenericitys(handlers[0].getClass());
        if (genericitys.size() >= 2) {
            Key<I, O> key = new Key(genericitys.get(0), genericitys.get(1));
            IChannel<I, O> channel = channelMap.get(key);
            if (channel == null) {
                channel = new DefaultChannel();
                channelMap.put(key, channel);
            }

            for (IChannelHandler<I, O> handler : handlers) {
                channel.pipeline().append(handler);
            }
        } else {
            log.error("[register] un support handlers. handlers={}", handlers);
            throw new RuntimeException("un support handlers");
        }
        return this;
    }

    @Override
    public <I, O> void process(I in, O out) {
        Key<I, O> key = new Key(in.getClass(), out.getClass());
        if (Void.TYPE.equals(out)) {
            key = new Key(in.getClass(), Void.class);
            out = null;
        }

        IChannel<I, O> channel = channelMap.get(key);
        if (channel == null) {
            log.error("[process] no channel. in={}, out={}", in, out);
            throw new RuntimeException("un support channel");
        }

        channel.process(in, out);
    }



    private static class Key<I, O> {

        private Class<I> inType;

        private Class<O> outType;

        public Key(Class<I> inType, Class<O> outType) {
            this.inType = inType;
            this.outType = outType;
        }

        @Override
        public boolean equals(Object o) {
            if (o instanceof Key) {
                Key key = (Key) o;
                if (this.inType.equals(key.inType) && this.outType.equals(key.outType)) {
                    return true;
                }
            }
            return false;
        }

        @Override
        public int hashCode() {
            return Objects.hash(inType, outType);
        }
    }
}

简化后使用

IChannelDispatcher bootstrap = new DefaultChannelDispatcher();
bootstrap.register(handler...);

bootstrap.process(in, out);

这里简化的是项目中多种事件的处理。只需要定义一次IChannelDispatcher即可，不用对每一套事件处理都初始化一遍IChannel了。