初学Java NIO框架Netty(一) 之 Hello

听朋友介绍这个框架于是学习之

首先非常感谢下面这两篇博文，很大部分代码和注释都参考了他们

http://deep-fish.iteye.com/blog/1819647

http://blog.163.com/linfenliang@126/blog/static/127857195201210744448686/?suggestedreading

没有废话了，直接代码加注释

HelloServer 服务端

package com.wei.study;

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.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

/**
 * 
 * @author wangwei
 * Jun 7, 2013
 */
public class HelloServer {

	public static void main(String[] args) {
		/*
		 * 基于NIO的用于创建服务器端channel的channelFactory
		 * 这个类利用非阻塞I/O模式引入NIO，这样可以有效地向多个并发连接提供Channel 
		 * 
		 * 在实现类中传入了两个cached线程池
		 * 前面一个代表boss线程,后面一个代表worker线程
		 * 
		 * Boss线程： 
		 * 每个server服务器都会有一个boss线程，每绑定一个InetSocketAddress都会产生一个boss线程，比如：我们开启了两个服务器端口80和443，则我们会有两个boss线程。
		 * 一个boss线程在端口绑定后，会接收传进来的连接，一旦连接接收成功，boss线程会指派一个worker线程处理连接。 
		 *
		 * Worker线程： 
		 * 一个NioServerSocketChannelFactory会有一个或者多个worker线程。一个worker线程在非阻塞模式下为一个或多个Channels提供非阻塞 读或写 
		 * 
		 * 简单来说，Boss线程用来接收客户端请求；worker用来执行非阻塞的IO操作，主要是read，write。
		 */
		ChannelFactory tcpChannelFactory =  new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),Executors.newCachedThreadPool());
		
		//server服务启动器
		ServerBootstrap bootstrap = new ServerBootstrap(tcpChannelFactory);
		
		//管道工厂 设置处理客户端消息和消息事件的Handler类
		bootstrap.setPipelineFactory(new ChannelPipelineFactory(){
			@Override
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new HelloServerHandler());
			}
		});
		//开放监听8000端口代客户端通信
		bootstrap.bind(new InetSocketAddress(8000));
	}
}

HelloServerHandler服务处理类

package com.wei.study;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

/**
 * 
 * @author wangwei
 * Jun 7, 2013
 */
public class HelloServerHandler extends SimpleChannelHandler {

	@Override
	//设置连接时触发事件
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
			throws Exception {
		System.out.println("Hello,I'm server.");
	}
	
	
}

HelloClient 客户端类

package com.wei.study;

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

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

/**
 * 
 * @author wangwei
 * Jun 7, 2013
 */
public class HelloClient {
	
	public static void main(String[] args) {
		//与服务端相同
		ChannelFactory clientSocketChannel = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),Executors.newCachedThreadPool());
		//与服务端相同
		ClientBootstrap bootstrap = new ClientBootstrap(clientSocketChannel);
		
		//与服务端相同
		bootstrap.setPipelineFactory(new ChannelPipelineFactory(){

			@Override
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new HelloClientHandler());
			}
		});
		
		//连接到IP地址及端口
		bootstrap.connect(new InetSocketAddress("127.0.0.1",8000));
	}
	
}

HelloClient 客户端处理类

package com.wei.study;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

/**
 * 
 * @author wangwei
 * Jun 7, 2013
 */
public class HelloClientHandler extends SimpleChannelHandler {
	
	@Override
	//与服务端相同
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
			throws Exception {
		System.out.println("Hello World,I'm client");
	}
	
}

-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------

最后符上前面提到的一篇博文的原文

Netty多线程处理机制

技术点描述

本文主要研究NioServerSocketChannelFactory类和NioDatagramChannelFactory类,

以及这两个类的各自作用。

由于基于pipelineFactory的线程池的配置已在前一节（Netty中execution包功能详解 ）中做了详细说明，而channelFactory中的线程池的配置并未做说明，本文档重点阐述channelFactory的线程池配置。NioServerSocketChannelFactory主要应用于TCP协议的数据处理，NioDatagramChannelFactory主要应用于UDP协议的数据处理。

实现方案

简单介绍一下实现该功能，使用了哪些技术或知识点

参考源码包

以下是对这两个类的具体说明以及相关类，重要的常用的方法的说明

1. NioServerSocketChannelFactory

此类常用于创建TCP的ServerBootstrap时作为构造方法的参数使用。

调用方式及设置如下：

ChannelFactory TCPCHANNEL_FACTORY = new NioServerSocketChannelFactory(

Executors.newCachedThreadPool(), Executors.newCachedThreadPool());

ServerBootstrap TCPSERVER_BOOTSTRAP = new ServerBootstrap(TCPCHANNEL_FACTORY);

基于NIO的用于创建服务器端channel的channelFactory，这个类利用非阻塞I/O模式引入NIO，这样可以有效地向多个并发连接提供Channel

线程的工作模式：

在此类中，有两种类型的线程，一种是boss线程，另一种是worker线程

Boss线程：

每个server服务器都会有一个boss线程，每绑定一个InetSocketAddress都会产生一个boss线程，比如：我们开启了两个服务器端口80和443，则我们会有两个boss线程。一个boss线程在端口绑定后，会接收传进来的连接，一旦连接接收成功，boss线程会指派一个worker线程处理连接。

Worker线程：

一个NioServerSocketChannelFactory会有一个或者多个worker线程。一个worker线程在非阻塞模式下为一个或多个Channels提供非阻塞 读或写

线程的生命周期和优雅的关闭

在NioServerSocketChannelFactory被创建的时候，所有的线程都会从指定的Executors中获取。Boss线程从bossExecutor中获取，worker线程从workerExecutor中获取。因此，我们应该准确的指定Executors可以提供足够数量的线程，最好的选择就是指定一个cached线程池（It is the best bet to specify a cached thread pool）。

此处发现所有源码中的例子(example)中均设置为Executors.newCachedThreadPool()

Boss线程和worker线程都是懒加载，没有程序使用的时候要释放掉。当boss线程和worker线程释放掉的时候，所有的相关资源如Selector也要释放掉。因此，如果想要优雅的关闭一个服务，需要做一下事情：

1. 对factory创建的channels执行解绑（unbind）操作
2. 关闭所有的由解绑的channels处理的子channels（这两步目前通常通过ChannelGroup.close()来操作）
3. 调用releaseExternalResources()方法

请确保在所有的channels都关闭前不要关闭executor，否则，会报RejectedExecutionException异常而且相关资源可能不会被释放掉。

在此类中，最关键的构造方法：

public NioServerSocketChannelFactory(

Executor bossExecutor, WorkerPool<NioWorker> workerPool) {

if (bossExecutor == null) {

throw new NullPointerException("bossExecutor");

}

if (workerPool == null) {

throw new NullPointerException("workerPool");

}

this.bossExecutor = bossExecutor;

this.workerPool = workerPool;

sink = new NioServerSocketPipelineSink(workerPool);

}

参数说明：

bossExecutor:默认推荐设置为Executors.newCachedThreadPool()

workerPool：可通过new NioWorkerPool(workerExecutor, workerCount, true)创建，workerExecutor默认推荐设置为Executors.newCachedThreadPool()，workerCount可设置为Runtime.getRuntime().availableProcessors() * 2（默认）

关于其中的NioWorkerPool类和NioServerSocketPipelineSink类，将在下文给予说明。

1. NioWorkerPool

此类的关键方法是super的AbstractNioWorkerPool，

构造方法：

AbstractNioWorkerPool(Executor workerExecutor, int workerCount, boolean allowShutDownOnIdle) {

…

}

参数说明：

workerExecutor：worker线程的Executor

workerCount：最多创建的worker线程数

allowShutDownOnldle：空闲时是否关闭线程

1. NioServerSocketPipelineSink

内部组件，传输服务的一个实现类，大多数情况下，不对用户开放。

1. NioDatagramChannelFactory

此类常用于创建UDP的ServerBootstrap时作为构造方法的参数使用。

调用方式及设置如下：

ChannelFactory UDPCHANNEL_FACTORY = new NioDatagramChannelFactory()

ConnectionlessBootstrap UDPSERVER_BOOTSTRAP = new ConnectionlessBootstrap(UDPCHANNEL_FACTORY);

这个类利用非阻塞I/O模式引入NIO，这样可以有效地向多个并发连接提供Channel

此类仅有一种线程类型：worker线程。一个NioDatagramChannelFactory可创建一个或多个worker线程，一个worker线程在非阻塞模式下为一个或多个Channels提供非阻塞 读或写

在NioDatagramChannelFactory被创建的时候，所有的线程都会从指定的Executors中获取

因此，我们应该准确的指定Executors可以提供足够数量的线程，最好的选择就是指定一个cached线程池（It is the best bet to specify a cached thread pool）。

所有的worker线程都是懒加载，没有程序使用的时候要释放掉。当worker线程释放掉的时候，所有的相关资源如Selector也要释放掉。因此，如果想要优雅的关闭一个服务，需要做一下事情：

通常通过ChannelGroup.close()来关闭所有的由此factory创建的channels

调用releaseExternalResources()方法

请确保在所有的channels都关闭前不要关闭executor，否则，会报RejectedExecutionException异常而且相关资源可能不会被释放掉。

不支持多播模式，如果需要多播模式支持，采用OioDatagramChannelFactory替代

以下是此类中的最重要的构造方法：

public NioDatagramChannelFactory(WorkerPool<NioDatagramWorker> workerPool, InternetProtocolFamily family) {

…

}

参数说明：

workerPool：参考NioServerSocketChannelFactory

family：网络协议系列，这个参数是为了UDP的多播模式的，此参数只在java7+版本才有效。默认为null，可通过StandardProtocolFamily.INET(Ipv4)或者StandardProtocolFamily.INET6(Ipv6)设置

1. NioClientSocketChannelFactory

与NioServerSocketChannelFactory类似，不做具体解说

评论

1 楼 1058106015 2014-08-07  

噢耶，看到我的博客被引用了