Channel的创建
前文中提到ServerBootStrap在启动配置中的方法channel（NioServerSocketChannel.class）会创建一个ReflectiveChannelFactory并赋值给成员变量channelFactory。当调用绑定端口号方法bind(8899)时，本质是通过反射调用NioServerSocketChannel的构造函数来创建Channel。
- 创建ChannelFactory实例：
public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}
- NioServerSocketChannel的构造函数
public class NioServerSocketChannel extends AbstractNioMessageChannel
implements io.netty.channel.socket.ServerSocketChannel {
private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
private static ServerSocketChannel newSocket(SelectorProvider provider) {
try {
return provider.openServerSocketChannel();
} catch (IOException e) {
throw new ChannelException(
"Failed to open a server socket.", e);
}
}
// 1. 构造函数，调用静态方法创建ServerSocketChannel
public NioServerSocketChannel() {
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
// 2. 调用父类构造函数
public NioServerSocketChannel(ServerSocketChannel channel) {
super(null, channel, SelectionKey.OP_ACCEPT);
config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
}
在构造函数中首先调用静态方法
newSocket()，创建一个Java Nio ServerSocketChannel实例，它继承SelectableChannel。-
然后调用父类
AbstractNioChannel的构造函数，它会持有上一步创建的ServerSocketChannel实例并赋值给变量ch，同时配置为非阻塞ch.configureBlocking(false)，如下：protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); // 1. NioServerSocketChannle(netty)持有ServerSocketChannel(jdk) this.ch = ch; this.readInterestOp = readInterestOp; try { // 2. 配置为非阻塞 ch.configureBlocking(false); } catch (IOException e) { try { ch.close(); } catch (IOException e2) { if (logger.isWarnEnabled()) { logger.warn( "Failed to close a partially initialized socket.", e2); } } throw new ChannelException("Failed to enter non-blocking mode.", e); } }
ChannelPipeline
1. ChannelPipeline的创建
继续查看父类AbstractChannel可以发现，在构造函数中，创建了ChannelPipeline，并赋值给pipeline对象。
protected AbstractChannel(Channel parent, ChannelId id) {
this.parent = parent;
this.id = id;
unsafe = newUnsafe();
// 1. 赋值给pipeline对象
pipeline = newChannelPipeline();
}
protected DefaultChannelPipeline newChannelPipeline() {
// 当前对象this，即Channel传入
return new DefaultChannelPipeline(this);
}
-
Channel通过变量
pipeline持有ChannelPipeline。 -
DefaultChannelPipeline的构造函数接受this为参数，因此ChannelPipeline也持有Channel。
2. ChannelPipeline的拦截过滤器模式
-
ChannelPipeline中可以添加多个
ChannelHandler，I/O事件在ChannelPipeline中依次传递。它也提供了添加、删除ChannelHandler的方法，如addLast()、removeLast()等。
ChannelPipeline是线程安全的，可以随时添加或删除ChannelHandler。
-
ChannelPipeline中的ChannelHandler分为InboundHandler和OutBoundHandler：InboundHandler只处理I/O输入请求，OutBoundHandler只处理I/O输出请求。例如：ChannelPipeline p = ...; p.addLast("1", new InboundHandlerA()); p.addLast("2", new InboundHandlerB()); p.addLast("3", new OutboundHandlerA()); p.addLast("4", new OutboundHandlerB()); p.addLast("5", new InboundOutboundHandlerX());
1至5是ChannelHandler的整体添加顺序，其中1和2是InboundHandler，3和4是OutboundHandler，而5即可以处理输入又可以处理输出。因此I/O事件的传播如上图：
输入: 1 -> 2 -> 5
输出: 5 -> 4 -> 3
-
ChannelPipeline中I/O事件的传播依靠调用ChannelHandlerContext的方法：
| Inbound | Outbound |
|---|---|
| fireChannelRegistered | bind |
| fireChannelActive | connect |
| fireChannelRead | write |
| fireExceptionCaught | flush |
| ...... | ...... |
public class MyInboundHandler extends {@link ChannelInboundHandlerAdapter} {
@Override
public void channelActive({@link ChannelHandlerContext} ctx) {
System.out.println("Connected!");
ctx.fireChannelActive();
}
}
在上例中，自定义Handler MyInboundHandler在自己的channelActive方法中调用了ctx.fireChannelActive()方法，将事件传播给下一个Handler。
3. ChannelPipeline中添加耗时任务的方式
当自定义Handler中需要处理耗时较长的任务时，有2种方式：
-
添加到
ChannelPipeline时，指定事件执行组EventExecutorGroup：EventExecutorGroup group = new DefaultEventExecutorGroup(16); ChannelPipeline pipeline = ......; pipeline.addLast("decoder", new MyProtocolDecoder()); pipeline.addLast("encoder", new MyProtocolEncoder()); // 指定事件执行组 pipeline.addLast(group, "handler", new MyBusinessLogicHandler()); 或者在自定义Handler中使用线程池ExecutorService处理。
