比较版本：Netty 3.10.5 和 Netty 4.1.17
总结： 3 和 4 比较起来，有两点印象很深刻：
一个是 4 没有了各种event 对象的创建，GC更友好。
另一个是 4的Handler 链总是IO 线程来处理，而3 是哪种线程发起了事件，哪种线程去处理Handler 链。
ServerBootstrap bootstrap = new ServerBootstrap();
//boss线程监听端口，worker线程负责数据读写
ExecutorService boss = Executors.newCachedThreadPool();
ExecutorService worker = Executors.newCachedThreadPool();
//设置niosocket工厂
bootstrap.setFactory(new NioServerSocketChannelFactory(boss, worker));
ExecutionHandler executionHandler =
new ExecutionHandler(
new OrderedMemoryAwareThreadPoolExecutor(2, 1048576, 1048576));
//设置管道的工厂
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() throws Exception {
return Channels.pipeline(new StringDecoder(),
new StringEncoder(), executionHandler, new HelloHandler());
}
});
bootstrap.bind(new InetSocketAddress(10102));
我们直接用上面 pipeline 里注册的handler 来举例分析：
- I/O线程（Work线程）将消息读取到SocketChannel的接收缓冲区中之后
- I/O线程生成相应的事件，触发事件执行。
- I/O线程调度执行ChannelPipeline中Handler链的对应方法。
- 执行到 executionHandler 的时候， 我们发现 任务消息已经投递到 业务线程池（ExecutionHandler）中了，之后的Handler的处理，其实就由业务线程来处理了。ExecutionHandler 的handleUpstream函数如下所示：
public void handleUpstream(
ChannelHandlerContext context, ChannelEvent e) throws Exception {
if (handleUpstream) {
executor.execute(new ChannelUpstreamEventRunnable(context, e, executor));
} else {
context.sendUpstream(e);
}
}
二、Netty3 写消息处理（业务线程触发的写操作）
- Netty将写操作封装成写事件传播。
//org.jboss.netty.channel.AbstractChannel#write(java.lang.Object)
public ChannelFuture write(Object message) {
return Channels.write(this, message);
}
// org.jboss.netty.channel.Channels#write(org.jboss.netty.channel.Channel, java.lang.Object, java.net.SocketAddress)
public static ChannelFuture write(Channel channel, Object message, SocketAddress remoteAddress) {
ChannelFuture future = future(channel);
// 在这创建了一个事件对象
channel.getPipeline().sendDownstream(
new DownstreamMessageEvent(channel, future, message, remoteAddress));
return future;
}
- 写事件被调度到ChannelPipeline中，由当前线程（当前是业务线程）按照Handler链串行调用支持Downstream事件的Handler。
- 执行到最后Handler的时候，将编码后的消息Push到发送队列（org.jboss.netty.channel.local.DefaultLocalChannel#writeBuffer）中，同时向 taskQueue 中加入 WriteTask，业务线程返回。
org.jboss.netty.channel.DefaultChannelPipeline.DefaultChannelHandlerContext#sendDownstream
public void sendDownstream(ChannelEvent e) {
DefaultChannelHandlerContext prev = getActualDownstreamContext(this.prev);
if (prev == null) {
try {
getSink().eventSunk(DefaultChannelPipeline.this, e);
} catch (Throwable t) {
notifyHandlerException(e, t);
}
} else {
DefaultChannelPipeline.this.sendDownstream(prev, e);
}
}
// org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink#eventSunk
public void eventSunk(
ChannelPipeline pipeline, ChannelEvent e) throws Exception {
Channel channel = e.getChannel();
if (channel instanceof NioServerSocketChannel) {
handleServerSocket(e);
} else if (channel instanceof NioSocketChannel) {
// 从这里投递消息 看发现像是绕了一个小圈，哈哈哈哈
handleAcceptedSocket(e);
}
}
// 但是这次不是 ChannelStateEvent 事件了，是messageEvent
private static void handleAcceptedSocket(ChannelEvent e) {
if (e instanceof ChannelStateEvent) {
……
} else if (e instanceof MessageEvent) {
MessageEvent event = (MessageEvent) e;
NioSocketChannel channel = (NioSocketChannel) event.getChannel();
// 数据写入 队列
boolean offered = channel.writeBufferQueue.offer(event);
assert offered;
channel.worker.writeFromUserCode(channel);
}
}
@Override
public void writeFromUserCode(final AbstractNioChannel channel) {
……
// 以当前的环境，会走这里，向taskQUeue 中注册writeTask
if (scheduleWriteIfNecessary(channel)) {
return;
}
……
}
@Override
protected boolean scheduleWriteIfNecessary(final AbstractNioChannel channel) {
final Thread currentThread = Thread.currentThread();
final Thread workerThread = thread;
if (currentThread != workerThread) {
if (channel.writeTaskInTaskQueue.compareAndSet(false, true)) {
// 看到这里其实就很熟悉了
registerTask(channel.writeTask);
}
return true;
}
return false;
}
- I/O线程从发送消息队列中取出消息，调用SocketChannel的write方法进行消息发送。
private static EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(2);
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline ph = ch.pipeline();
// 以("n")为结尾分割的 解码器
ph.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 解码和编码，应和客户端一致
ph.addLast("decoder", new StringDecoder());
ph.addLast("encoder", new StringEncoder());
ph.addLast(businessGroup, new NettyServerHandler());// 服务端业务逻辑
}
}
把业务线程池和 Handler 绑定，这一块和3不太一样了。
-
I/O线程NioEventLoop从SocketChannel中读取数据报，将ByteBuf投递到ChannelPipeline，触发ChannelRead事件。
-
I/O线程NioEventLoop调用ChannelHandler链，直到将消息投递到业务线程，然后I/O线程返回。
// io.netty.channel.DefaultChannelPipeline#fireChannelRead
@Override
public final ChannelPipeline fireChannelRead(Object msg) {
AbstractChannelHandlerContext.invokeChannelRead(head, msg);
return this;
}
// io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead(io.netty.channel.AbstractChannelHandlerContext, java.lang.Object)
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
// 得到Handler 执行的线程组
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeChannelRead(m);
} else {
// 具体业务交给业务线程
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelRead(m);
}
});
}
}
三、Netty4 写消息处理（业务线程发起的写操作）
- 将发送消息封装成Task，放入到I/O线程NioEventLoop的任务队列中，由NioEventLoop在循环中统一调度和执行。放入任务队列之后，业务线程返回；
private void write(Object msg, boolean flush, ChannelPromise promise) {
AbstractChannelHandlerContext next = findContextOutbound();
final Object m = pipeline.touch(msg, next);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
if (flush) {
next.invokeWriteAndFlush(m, promise);
} else {
next.invokeWrite(m, promise);
}
} else {
// 业务线程触发的事件 封装了一个task
AbstractWriteTask task;
if (flush) {
task = WriteAndFlushTask.newInstance(next, m, promise);
} else {
task = WriteTask.newInstance(next, m, promise);
}
safeExecute(executor, task, promise, m);
}
}
private static void safeExecute(EventExecutor executor, Runnable runnable, ChannelPromise promise, Object msg) {
try {
executor.execute(runnable);
} catch (Throwable cause) {
try {
promise.setFailure(cause);
} finally {
if (msg != null) {
ReferenceCountUtil.release(msg);
}
}
}
}
// io.netty.util.concurrent.SingleThreadEventExecutor#execute
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
startThread();
// 添加到taskQUeue 中
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
protected void addTask(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
if (!offerTask(task)) {
reject(task);
}
}
final boolean offerTask(Runnable task) {
if (isShutdown()) {
reject();
}
return taskQueue.offer(task);
}
- I/O线程NioEventLoop调用ChannelHandler链，进行消息发送，处理Outbound事件，直到将消息放入发送队列，然后唤醒Selector，进而执行写操作。
参考文章：netty3与netty4的区别
