责任链模式
2023-07-09 本文已影响0人
给代码点颜色
写在前面的话
很多的系统在不断迭代的过程中，免不了代码的默认规范越来越不可控，越来越堆积，这里希望近一点绵薄之力，希望既能相对规范下代码的逻辑，又可以方便后来者更好的融入前人的规范中，并轻松的进行扩展和优化。
注：
相关系列文章会不定期更新（主要因为我比较懒），有想法的读者可以留言探讨哈~
另外，如有引用或者拷贝的请注明出处。
废话不多说了，下面进入新的废话中~
责任链模式
思考部分
一旦说道责任链模式，我们能想到的需要的组件有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了。
