netty ExecutionHandler

前面(http://san-yun.iteye.com/blog/1693598)提到netty通过ChannelFuture来提供异步IO，但我实际测试中发现并不是总是这样的，需要ChannelPipelineFactory中指定ExecutionHandler。测试代码：

 

@ChannelPipelineCoverage("all")
public class DiscardServerHandler extends SimpleChannelHandler {

	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		Channel channel =e.getChannel();
		ChannelFuture future =  channel.write(e.getMessage());
		future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Thread.sleep(1000);  
                System.out.println(Thread.currentThread().getName());
                System.out.println("ok1");
            }
        });
		System.out.println(Thread.currentThread().getName());
		System.out.println("ok2");
	}

 

 

package com.duitang.test.test;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;


public class NettyTest {

    public static void main(String[] args) throws SecurityException, Exception {
        
        ChannelFactory factory =  
                new NioServerSocketChannelFactory (  
                        Executors.newCachedThreadPool(),  
                        Executors.newCachedThreadPool());  
        
        ServerBootstrap bootstrap = new ServerBootstrap (factory);
        DiscardServerHandler handler = new DiscardServerHandler();
        ChannelPipeline pipeline = bootstrap.getPipeline();  
        pipeline.addLast("handler", handler);  
        bootstrap.setOption("child.tcpNoDelay", true);  
        bootstrap.setOption("child.keepAlive", true);  
        bootstrap.bind(new InetSocketAddress(8080));
        
    }

}

 

输出ok1，ok2，同步执行。

 

public static void main(String[] args) {
		ChannelFactory factory = new NioServerSocketChannelFactory(
				Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool());
		ServerBootstrap bootstrap = new ServerBootstrap(factory);
		ExecutionHandler executionHandler = new ExecutionHandler(
				new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576));
		DatabaseGatewayPipelineFactory gatewayPipelineFactory = new DatabaseGatewayPipelineFactory(
				executionHandler);
		bootstrap.setPipelineFactory(gatewayPipelineFactory);
		bootstrap.setOption("child.tcpNoDelay", true);
		bootstrap.setOption("child.keepAlive", true);
		bootstrap.bind(new InetSocketAddress(8080));
	}

	static class DatabaseGatewayPipelineFactory implements
			ChannelPipelineFactory {

		private final ExecutionHandler executionHandler;

		public DatabaseGatewayPipelineFactory(ExecutionHandler executionHandler) {
			this.executionHandler = executionHandler;
		}

		public ChannelPipeline getPipeline() throws Exception {
			ChannelPipeline pipeline = Channels.pipeline();
			pipeline.addFirst("decoder", new DiscardServerHandler());
			pipeline.addFirst("encoder", new DiscardServerHandler());
			pipeline.addFirst("executor", executionHandler);
			pipeline.addLast("handler", new DiscardServerHandler());
			return pipeline;
		}

	}

 输出

pool-3-thread-2
ok2
New I/O server worker #1-1
ok1