import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
public static ChannelHandler createDecoder() {
	int maxObjectSize = AppProperties.getAsInt("rpc_common_maxObjectSize",
			1024 * 1024);
	String f = AppProperties.get("rpc_common_codec", "java");
	if (f.equalsIgnoreCase("java")) {
		return new ObjectDecoder(maxObjectSize,
				ClassResolvers.softCachingConcurrentResolver(null));
	} else if (f.equalsIgnoreCase("jbossSerialization")) {
		return new JBossSerializationDecoder(maxObjectSize);
	} else if (f.equalsIgnoreCase("json")) {
		throw new AppRuntimeException("not support " + f + " yet!");
	} else if (f.equalsIgnoreCase("hessian")) {
		return new HessianDecoder(maxObjectSize);
	} else {
		throw new AppRuntimeException("not support " + f + " yet!");
	}
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
/**
 * Start rpc server which is used to update progress.
 */
private void startRPCServer() {
    this.rpcServer = new ServerBootstrap(new NioServerSocketChannelFactory(
            Executors.newFixedThreadPool(GuaguaYarnConstants.DEFAULT_STATUS_RPC_SERVER_THREAD_COUNT),
            Executors.newCachedThreadPool(new MasterThreadFactory())));

    // Set up the pipeline factory.
    this.rpcServer.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new ObjectEncoder(),
                    new ObjectDecoder(ClassResolvers.cacheDisabled(getClass().getClassLoader())),
                    new ServerHandler());
        }
    });

    // Bind and start to accept incoming connections.
    this.rpcServer.bind(new InetSocketAddress(rpcPort));
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
/**
 * Connect to app master for status RPC report.
 */
private void initRPCClient() {
    this.rpcClient = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newSingleThreadExecutor(),
            Executors.newSingleThreadExecutor()));

    // Set up the pipeline factory.
    this.rpcClient.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new ObjectEncoder(),
                    new ObjectDecoder(ClassResolvers.cacheDisabled(getClass().getClassLoader())),
                    new ClientHandler());
        }
    });

    // Start the connection attempt.
    ChannelFuture future = this.rpcClient.connect(new InetSocketAddress(this.rpcHostName, this.rpcPort));
    LOG.info("Connect to {}:{}", this.rpcHostName, this.rpcPort);
    this.rpcClientChannel = future.awaitUninterruptibly().getChannel();
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
/**
 * Creates a new instance that talks to a {@link DefaultServer} on the specified host and port.
 * 
 * @param clientId
 *            A unique id. No two clients with the same id can access the same
 *            {@link DefaultServer}
 * @param host
 *            The host the {@link DefaultServer} runs on
 * @param port
 *            The port the {@link DefaultServer} runs on
 */
public DefaultClient(final String clientId, final String host, final int port) {
	this.clientId = clientId;
	this.host = host;
	this.port = port;

	// Configure the client.
	bootstrap = new ClientBootstrap(
			new OioClientSocketChannelFactory(Executors.newCachedThreadPool(new DaemonThreadFactory())));

	// Set up the pipeline factory.
	bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
		@Override
		public ChannelPipeline getPipeline() throws Exception {
			return Channels.pipeline(
					new ObjectEncoder(ENCODER_ESTIMATED_LENGTH),
					new ObjectDecoder(DECODER_ESTIMATED_LENGTH, ClassResolvers.weakCachingResolver(null)),
					new ClientHandshakeHandler(new Handshake(clientId), HANDSHAKE_TIMEOUT_MILLIS),
					clientHandler);
		}
	});

	bootstrap.setOption("tcpNoDelay", true);
	bootstrap.setOption("keepAlive", true);
	bootstrap.setOption("soTimeout", 10000L);
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
public static ChannelHandlerFactory newObjectDecoder() {
    return new ChannelHandlerFactory() {
        @Override
        public ChannelHandler newChannelHandler() {
            return new ObjectDecoder(ClassResolvers.weakCachingResolver(null));
        }
    };
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
@Override
public ChannelPipeline getPipeline() throws Exception
{
	ChannelPipeline pipeline = Channels.pipeline();
	// ���������
	pipeline.addLast("objectencoder", new ObjectEncoder());
	// ���������
	pipeline.addLast("objectdecoder", new ObjectDecoder(ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
	// ��������ҵ����
	pipeline.addLast("clientHandler", myClientHandler);
	return pipeline;
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
@Override
public ChannelPipeline getPipeline() throws Exception {
	ChannelPipeline pipeline = Channels.pipeline();

	pipeline.addLast("objectencoder", new ObjectEncoder());

	pipeline.addLast("objectdecoder", new ObjectDecoder(ClassResolvers
			.cacheDisabled(this.getClass().getClassLoader())));
	pipeline.addLast("serverhandler", new MyServerHandler());
	return pipeline;
}
 

import org.jboss.netty.handler.codec.serialization.ClassResolvers; //导入依赖的package包/类
/**
 * Creates a new server instance on the specified port.
 * 
 * @param port
 *            The port
 */
public DefaultServer(final int port) {
	this.port = port;

	// Configure the server.
	bootstrap = new ServerBootstrap(
			new OioServerSocketChannelFactory(
					Executors.newCachedThreadPool(new DaemonThreadFactory()),
					Executors.newCachedThreadPool(new DaemonThreadFactory())));

	// Set up the pipeline factory.
	bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
		@Override
		public ChannelPipeline getPipeline() throws Exception {
			return Channels.pipeline(
					new ObjectEncoder(ENCODER_ESTIMATED_LENGTH), // 1 MB default size
					new ObjectDecoder(DECODER_ESTIMATED_LENGTH, ClassResolvers.weakCachingResolver(null)), // 20 MB max. size - should be largely sufficient
					new ServerHandshakeHandler(channelContainer, HANDSHAKE_TIMEOUT_MILLIS),
					serverHandler);
		}
	});

	bootstrap.setOption("tcpNoDelay", true);
	bootstrap.setOption("keepAlive", true);
	bootstrap.setOption("child.tcpNoDelay", true);
	bootstrap.setOption("child.keepAlive", true);
}