Netty学习

艾伦蓝

浏览: 596327 次
性别:
来自: 厦门

最近访客更多访客>>

stephen830

njdccy

lzy8828

WangJiaX

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

Netty

Netty是一个基于JAVA NIO类库的异步通信框架，它的架构特点是：异步非阻塞、基于事件驱动、高性能、高可靠性和高可定制性。
在Netty中Client和Server通过Channel连接，然后通过ByteBuffer进行传输。每个Channel有自己的Pipeline，Pipeline上面可以添加和定义Handler和Event。

Netty 4.0中，定义了Channel接口，这个接口用于连接网络的socket传输，或者具有I/O操作的组件连接。这里的I/O操作有，read,write,bind,connect.

Channel接口为用户提供了：

1.   Channel的当前状态，比如：Channel是否open，或者Channel是否已经连接。

2.   Channel的参数，比如：接受的buffer大小。

3.   Channel支持的I/O操作，比如：read,write,connect,bind。

4.   注册在Channel上的ChannelPipeline，ChannelPipeline用于处理所有的I/O事件和请求。

ChannelHandler接口用于处理和拦截Channel接口中的ChannelEvents。Netty中的ChannelPipeline概念使用了Intecepting Filter Patter模式来实现，这样做的好处是允许用户可以完全控制Event的处理，并且控制ChannelHandlers在ChannelPipeline的交互。

当一个全新的Channel创建的时候，都必须创建一个ChannelPipeline，并使Channel和ChannelPipeline相关联。这种关联关系是永久性的，这意味着，一旦一个ChannelPipeline和Channel关联了，那么这个Channel就再也无法关联其他ChannelPipeline，也无法取消与当前ChannelPipeline的关联。

【官方推荐】使用Channel接口中的pipeleine()方法来创建一个ChannelPipelien，而不要用new去实例一个ChannePipeline类

ChannelPipeline pipeline = Channels.pipeline();

Pipeline中的事件流

图中显示了一个典型的ChannelHandler或者ChannelPipeline对于ChannelEvent的处理流程。ChannelHandler接口有两个子类，分别是ChannelUpstreamHandler(ChannelInboundHandler)和ChannelDownstreamHandler(ChannelOutBoundstreamHandler)。这两个之类用于处理每一个ChannelEvent，然后由ChannelHandlerContext.sendUpstream(ChannelEvent)和ChannelHandlerContext.sendDownstream(ChannelEvent)将每一个ChannelEvent转发到最近的handler。根据upstream和downstream的不同，每个Event的处理也会有所不同。

在一条Pipeline中，一般会有一个或者多个ChannelHandler用于接收I/O事件(read)或者请求I/O操作(write，close)。一个典型的服务器会有如下的一个ChannelPipeline用于处理不同的ChannelHandler。

ChannelPipeline p = Channels.pipeline();

p.addLast(“1”, new UpstreamHandlerA());

p.addList(“2”, new UpstreamHandlerB());

p.addList(“3”, new DownstreamHandlerA());

p.addList(“4”, new DownstreamHandlerB());

p.addList(“5”, new UpstreamHandlerX());

第一个Netty例子：

服务端

/**
 * Netty 服务端代码
 * 
 * @author lihzh
 * @alia OneCoder
 * @blog http://www.coderli.com
 */
public class HelloServer {

	public static void main(String args[]) {
		// Server服务启动器
		ServerBootstrap bootstrap = new ServerBootstrap(
				new NioServerSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		// 设置一个处理客户端消息和各种消息事件的类(Handler)
//两个线程池一个用于监听请求，并分派给slave进行处  一个用于处理请求，将其丢到线程池中处
		bootstrap
				.setPipelineFactory(new ChannelPipelineFactory() {
					@Override
					public ChannelPipeline getPipeline()
							throws Exception {
						return Channels
								.pipeline(new HelloServerHandler());
					}
				});
		// 开放8000端口供客户端访问。
		bootstrap.bind(new InetSocketAddress(8000));
	}

	private static class HelloServerHandler extends
			SimpleChannelHandler {

		/**
		 * 当有客户端绑定到服务端的时候触发，打印"Hello world, I'm server."
		 * 
		 * @alia OneCoder
		 * @author lihzh
		 */
		@Override
		public void channelConnected(
				ChannelHandlerContext ctx,
				ChannelStateEvent e) {
			System.out.println("Hello world, I'm server.");
		}
	}
}

客户端

/**
 * Netty 客户端代码
 * 
 * @author lihzh
 * @alia OneCoder
 * @blog http://www.coderli.com
 */
public class HelloClient {

	public static void main(String args[]) {
		// Client服务启动器
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		// 设置一个处理服务端消息和各种消息事件的类(Handler)
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			@Override
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new HelloClientHandler());
			}
		});
		// 连接到本地的8000端口的服务端
		bootstrap.connect(new InetSocketAddress(
				"127.0.0.1", 8000));
	}

	private static class HelloClientHandler extends SimpleChannelHandler {


		/**
		 * 当绑定到服务端的时候触发，打印"Hello world, I'm client."
		 * 
		 * @alia OneCoder
		 * @author lihzh
		 */
		@Override
		public void channelConnected(ChannelHandlerContext ctx,
				ChannelStateEvent e) {
			System.out.println("Hello world, I'm client.");
		}
	}
}

Server端开放端口，供Client连接，Client发起请求，连接到Server指定的端口，完成绑定。随后便可自由通信。其实就是普通Socket连接通信的过程。
Netty框架是基于事件机制的，简单说，就是发生什么事，就找相关处理方法。就跟着火了找119，生病了找120一个道理。

Server

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;


public class HelloNettyClient {

	public static void main(String[] args) {
		// Configure the client.
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));

		// Set up the default event pipeline.
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			@Override
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new StringDecoder(),
						new StringEncoder(), new ClientHandler());
			}
		});

		// Start the connection attempt.
		ChannelFuture future = bootstrap.connect(new InetSocketAddress(
				"localhost", 8000));

		// Wait until the connection is closed or the connection attempt fails.
		future.getChannel().getCloseFuture().awaitUninterruptibly();

		// Shut down thread pools to exit.
		bootstrap.releaseExternalResources();
	}

	private static class ClientHandler extends SimpleChannelHandler {
		private BufferedReader sin = new BufferedReader(new InputStreamReader(
				System.in));
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			if (e.getMessage() instanceof String) {
				String message = (String) e.getMessage();
				System.out.println(message);

				e.getChannel().write(sin.readLine());

				System.out.println("\n等待客户端输入。。。");
			}

			super.messageReceived(ctx, e);
		}

		@Override
		public void channelConnected(ChannelHandlerContext ctx,
				ChannelStateEvent e) throws Exception {
			System.out.println("已经与Server建立连接。。。。");
			System.out.println("\n请输入要发送的信息：");
			super.channelConnected(ctx, e);

			e.getChannel().write(sin.readLine());
		}
	}
};

Client

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;


public class HelloNettyClient {

	public static void main(String[] args) {
		// Configure the client.
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));

		// Set up the default event pipeline.
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			@Override
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new StringDecoder(),
						new StringEncoder(), new ClientHandler());
			}
		});

		// Start the connection attempt.
		ChannelFuture future = bootstrap.connect(new InetSocketAddress(
				"localhost", 8000));

		// Wait until the connection is closed or the connection attempt fails.
		future.getChannel().getCloseFuture().awaitUninterruptibly();

		// Shut down thread pools to exit.
		bootstrap.releaseExternalResources();
	}

	private static class ClientHandler extends SimpleChannelHandler {
		private BufferedReader sin = new BufferedReader(new InputStreamReader(
				System.in));
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			if (e.getMessage() instanceof String) {
				String message = (String) e.getMessage();
				System.out.println(message);

				e.getChannel().write(sin.readLine());

				System.out.println("\n等待客户端输入。。。");
			}

			super.messageReceived(ctx, e);
		}

		@Override
		public void channelConnected(ChannelHandlerContext ctx,
				ChannelStateEvent e) throws Exception {
			System.out.println("已经与Server建立连接。。。。");
			System.out.println("\n请输入要发送的信息：");
			super.channelConnected(ctx, e);

			e.getChannel().write(sin.readLine());
		}
	}
};