pipeline机制

最近在看netty的源码，准备将一些理解和总结写出来，netty的源码写的很漂亮理解起来也不是那么容易，很值得我们去学习和借鉴。我们知道在设计模式里面提到过一种责任链的模式，据我看一些源码的了解（不管是公司的中间件还是一些开源的项目），责任链的应用非常广泛，包括现在在公司做的东西。其中netty很好的扩展了责任链模式来实现netty的骨架，我们这里称它为pipeline模式。

如果将数据比作是水，那么pipeline就是水管，里面的阀门就是用来处理水的部分。比如一个Request进来，可能经过层层的处理，包括响应也是，所以类似这种服务端的框架很适合使用pipeline模式，实际上tomcat就是这么做的。netty的pipeline机制非常方便用户使用，定制自己的阀门，pipeline的结构如下图：



其中的HandlerContext就是上面讲的阀门，包括拦截器模式，Filter模式它们的实现都是类似的，全部是基于责任链的变体，每个HandlerContext都聚合一个Handler，由Handler负责具体的逻辑处理，netty中抽象出一个头和尾节点，其中从头向尾执行的时候，头部的逻辑不会执行，从尾向头执行的时候，尾部的逻辑不会执行，这两个虚节点也方便双向链表的实现。自己模拟netty的pipeline实现了一个简化版本的pipeline。下面来看一下代码：

package s4netty.pipeline;

/**
* pipeline接口，可以向pipeline中新增handler
*
* @author Administrator
*
*/
public interface Pipeline {

Pipeline addFirst(String name, Handler handler);

Pipeline fireProcess();
}可以向一个Pipeline中增加Handler，同时可以由Pipeline触发整个链的执行，这里只是一个简单的无参方法fireProcess。
package s4netty.pipeline;

/**
* handler接口
*
* @author Administrator
*
*/
public interface Handler {
void process(HandlerContext context);
}Handler接口定义，对应pipeline提供了process方法用来处理数据，其中需要的数据会存储在HandlerContext中。
package s4netty.pipeline;
/**
* 抽象handler执行上下文
* @author Administrator
*
*/
public interface HandlerContext {

Handler handler();

HandlerContext fireProcess();
}上下文的定义同样很简单，一个handler方法由子类实现去返回自己的Handler，同时fireProcess方法负责处理业务逻辑和流程的传递。到现在还只是一些简单接口的定义，可能还看不出来pipeline的端倪，不着急继续往下。
package s4netty.pipeline;

/**
* 抽象的handler上下文，基于责任链模式实现双向链表
*
* @author Administrator
*
*/
public abstract class AbstractHandlerContext implements HandlerContext {

// handler-context的前一个结点
AbstractHandlerContext prev;

// handler-context的下一个节点
AbstractHandlerContext next;

// handlerContext的名字
private final String name;

AbstractHandlerContext(String name) {
this.name = name;
}

public String getName() {
return name;
}

@Override
public HandlerContext fireProcess() {
// 获取下一个上下文，如果子类不复写该方法，则直接传递到下一个处理节点
AbstractHandlerContext next = this.next;
next.invokeProcess();
return this;
}

private void invokeProcess() {
handler().process(this);
}
}
抽象的Context是一个很关键的实现，因为它实际上是一个双向的链表，有next节点和prev节点，其中前面讲过fireProcess方法负责业务处理和责任链的传递，在这里也体现出来了。
package s4netty.pipeline;

/**
* 默认实现，关联一个Handler
*
* @author Administrator
*
*/
public class DefaultHandlerContext extends AbstractHandlerContext {

private Handler handler;

DefaultHandlerContext(String name, Handler handler) {
super(name);
this.handler = handler;
}

@Override
public Handler handler() {
return handler;
}

}
HandlerContext的默认实现，很简单，就是聚合了一个Handler。
package s4netty.pipeline;

import java.util.HashMap;
import java.util.Map;

/**
* pipeline模式
*
* @author Administrator
*
*/
public final class DefaultPipeline implements Pipeline {

final AbstractHandlerContext head;

final AbstractHandlerContext tail;

private Map<String, Handler> ctxHandlers = new HashMap<String, Handler>(4);

public DefaultPipeline() {

head = new HeadHandlerContext();
tail = new TailHandlerContext();

// 双向链表
head.next = tail;
tail.prev = head;
}

static final class HeadHandlerContext extends AbstractHandlerContext implements Handler {

private static final String HEADNAME = "HEAD";

HeadHandlerContext() {
super(HEADNAME);
}

@Override
public Handler handler() {
return this;
}

@Override
public void process(HandlerContext context) {
System.out.println(this.getName());
}

}

static final class TailHandlerContext extends AbstractHandlerContext implements Handler {

private static final String TAILNAME = "TAIL";

TailHandlerContext() {
super(TAILNAME);
}

@Override
public Handler handler() {
return this;
}

@Override
public void process(HandlerContext context) {
System.out.println(this.getName());
}

}

/**
* 实现pipeline接口中的addFirst方法，将handler置为head
*/
@Override
public Pipeline addFirst(String name, Handler handler) {

AbstractHandlerContext newCtx = new DefaultHandlerContext(name, handler);
AbstractHandlerContext nextCtx = head.next;
newCtx.prev = head;
newCtx.next = nextCtx;

head.next = newCtx;
nextCtx.prev = newCtx;

ctxHandlers.put(name, handler);
return this;
}

public Map<String, Handler> getCtxHandlers() {
return ctxHandlers;
}

public void setCtxHandlers(Map<String, Handler> ctxHandlers) {
this.ctxHandlers = ctxHandlers;
}

/**
* 触发处理流程
*/
@Override
public Pipeline fireProcess() {
//从头开始触发流程
head.fireProcess();
return this;
}

}
另外一个重要的类是Pipeline的实现，默认构建了一个只有头尾的双向链表，可以通过addFirst方法向其中添加其他的Handler。
以上就是pipeline机制的简单实现，整体上还是比较清晰的。