Netty -入门NIO初级编程

接上篇，今天主要讲解NIO的编程入门，借用上一篇的例子，这次试用NIO进行一次升级。

 

一、 TimeServer端，步骤如下

1、打开ServerSocketChannel，用于监听用户端的连接，他是所有客户端连接的父管道

2、绑定监听端口，设置连接为非阻塞模式

3、创建Reactor线程，创建多路复用器并启动线程

4、将ServerSocketChannel注册到Reactor线程多路复用器Selector上，监听ACCEPT事件

5、多路复用器在线程run方法的无限循环体内轮询准备就绪的Key

6、多路复用器监听到有新的客户端接入，处理新的接入请求，完成TCP三次握手，建立物理链路

7、设置客户端链路为非阻塞模式

8、将新接入的客户端连接注册到Reactor线程的多路复用器上，监听读操作，读取客户端发送的网络数据

9、异步读取客户端的请求消息到缓冲区

10、对ByteBuffer进行编解码，如果有板报消息指针reset，继续读取后续的报文，将解码成功的消息封装成Task，投递到业务线程池中，进行业务逻辑编排

11、将POJO对象encode成ByteBuffer，调用SocketChannel的异步write接口，将消息发送到客户端

 

示例代码：

package com.techstar.nio;

import java.io.IOException;

public class TimeServer {

	
	/**
	 * 启动服务
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		
		int port = 8081;
		
		MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
		
		new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
		System.out.println("TimeServer.started");
	}
}

 MultiplexerTimeServer

package com.techstar.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class MultiplexerTimeServer implements Runnable {

	private Selector selector;
	private ServerSocketChannel serverChannel;
	private volatile boolean stop;
	
	
	
	public MultiplexerTimeServer(int port) {
		super();
		try {
			this.selector = Selector.open();
			this.serverChannel = ServerSocketChannel.open();
			this.serverChannel.configureBlocking(false);
			this.serverChannel.socket().bind(new InetSocketAddress("127.0.0.1", port), 1024);
			
			this.serverChannel.register(this.selector, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
	}

	
	public void stop () {
		this.stop = true;
	}


	@Override
	public void run() {
		// TODO Auto-generated method stub
		while(!this.stop) {
			try {
				this.selector.select(1000);
				Set<SelectionKey> selectionKeys = this.selector.selectedKeys();
				Iterator<SelectionKey> iterator = selectionKeys.iterator();
				SelectionKey key = null;
				while (iterator.hasNext()) {
					key = iterator.next();
					iterator.remove();
					try {
						handleInput(key);
					}  catch (IOException e) {
						key.cancel();
						if(key.channel() != null) {
							key.channel().close();
						}
					}
				}
				
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			System.out.println("read end ...");
		}
		
		if (this.selector != null) {
			try {
				this.selector.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
	private void handleInput(SelectionKey key) throws IOException {
		if (key.isValid()) {
			if (key.isAcceptable()) {
				ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
				SocketChannel sc = ssc.accept();
				sc.configureBlocking(false);
				sc.register(this.selector, SelectionKey.OP_READ);
			}
			
			if (key.isReadable()) {
				SocketChannel sc = (SocketChannel) key.channel();
				
				ByteBuffer readBuffer = ByteBuffer.allocate(1024);
				int readBytes = sc.read(readBuffer);
				if (readBytes > 0) {
					readBuffer.flip();  //很重要， 在读或者写之前，必须将buffer flip
					byte[] bytes = new byte[readBuffer.remaining()];
					readBuffer.get(bytes);
					String body = new String(bytes, "UTF-8");
					
					System.out.println("The Time Server receive order : " + body);
					
					String currentTime = "QUERY TIME ORDER".equals(body)?new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
					
					this.doWrite(sc, currentTime);
				} else if (readBytes < 0) {
					key.cancel();
					sc.close();
				} else {
					System.out.println("没有收到任何数据");
				}
			}
		}
	}

	
	private void doWrite(SocketChannel sc, String response) throws IOException {
		
		if (response != null && response.trim().length() > 0) {
			System.out.println("send to client : " + response);
			byte[] bytes = response.getBytes();
			ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
			writeBuffer.put(bytes);
			writeBuffer.flip();  //很重要， 在读或者写之前，必须将buffer flip
			sc.write(writeBuffer);
		}
	}
}

 

 

 

二、TimeClient端

1、打开SocketChannel，绑定客户端地址（可选，默认系统会随机分配一个可用的本地地址）

2、设置SocketChannel为非线程阻塞模式

3、异步连接服务端

4、判断是否连接成功，如果连接成功则直接注册读状态位到多路复用器中，如果连接没成功（异步连接，返回false，说明客户端已经发送sync包，服务端没有返回ack包，物理连接还没有建立）

5、向Reactor线程的多路复用器注册OP_CONNECT状态位，监听服务器端的TCP的ACK应答

6、创建Reactor线程，创建多路复用器并启动线程

7、多路复用器在线程run方法的无限循环体内轮询准备就绪的Key

8、接收connect事件进行处理

9、判断连接结果，如果连接成功，注册读事件到多路复用器

10、异步读取客户端接收到的消息到缓冲区

11、对ByteBuffer进行解码

12、将POJO对象encode成ByteBuffer，调用SocketChannel的异步write接口，将消息发送给服务器端

示例代码

package com.techstar.nio;

public class TimeClient {

	public static void main(String[] args) {
		
		int port = 8081;
		
		new Thread(new TimeClientHandle(null, port), "TimeClient-001" ).start();;
	}
}

 

/**
 * 
 */
package com.techstar.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * @author mrh
 *
 */
public class TimeClientHandle implements Runnable {
	
	private String host;
	
	private int port;
	
	private Selector selector;
	
	private SocketChannel socketChannel;
	
	private volatile boolean stop;

	public TimeClientHandle(String host, int port) {
		super();
		this.host = host == null ? "127.0.0.1":host;
		this.port = port;
		
		try {
			this.selector = Selector.open();
			this.socketChannel = SocketChannel.open();
			this.socketChannel.configureBlocking(false);
		} catch (IOException e) {
			e.printStackTrace();
			System.exit(1);
		}
		
		
	}




	/* (non-Javadoc)
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run() {
		
		try {
			doConnect();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.exit(1);
		}
		
		
		while(!this.stop) {
			try {
				selector.select(1000);
				Set<SelectionKey> selectionKeys = this.selector.selectedKeys();
				
				Iterator<SelectionKey> iterator = selectionKeys.iterator();
				
				
				SelectionKey key= null;
				
				while (iterator.hasNext()) {
					key = iterator.next();
					iterator.remove();
					try {
						handleInput(key);
					} catch (IOException e) {
						if (key != null) {
							key.cancel();
							if (key.channel() != null) {
								key.channel().close();
							}
						}
					}
				}
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				System.exit(1);
			}
			System.out.println("wait.....");
		}
		
		if (this.selector != null) {
			try {
				this.selector.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	private void doConnect()  throws IOException {
		if (this.socketChannel.connect(new InetSocketAddress(this.host, port))) {
			this.socketChannel.register(selector, SelectionKey.OP_READ);
			this.doWrite(this.socketChannel);
			System.out.println("connected successed");
		} else {
			this.socketChannel.register(selector, SelectionKey.OP_CONNECT);
		}
	}

	
	private void handleInput(SelectionKey key) throws IOException{
		
		if (key.isValid()) {
			SocketChannel sc = (SocketChannel) key.channel();
			if (key.isConnectable()) {
				if (sc.finishConnect()) {
					sc.register(this.selector, SelectionKey.OP_READ);
					this.doWrite(sc);
				} else {
					System.exit(1);//连接失败
				}
			}
			
			if (key.isReadable()) {
				ByteBuffer readBuffer = ByteBuffer.allocate(1024);
				int readBytes = sc.read(readBuffer);
				if (readBytes > 0) {
					readBuffer.flip(); //很重要， 在读或者写之前，必须将buffer flip
					byte[] bytes = new byte[readBuffer.remaining()];
					readBuffer.get(bytes);
					
					String body = new String(bytes,"UTF-8");
					System.out.println(body);
					
					//重复请求
					this.doWrite(sc);
				} else if (readBytes < 0) {
					key.cancel();
					sc.close();
				} else {
					System.out.println("等待接收数据");
				}
			}
		}
	}
	
	private void doWrite(SocketChannel sc) throws IOException {
		String  response = "QUERY TIME ORDER";
		if (response != null && response.trim().length() > 0) {
			byte[] bytes = response.getBytes();
			ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
			writeBuffer.put(bytes);
			writeBuffer.flip();  //很重要， 在读或者写之前，必须将buffer flip
			sc.write(writeBuffer);
			if (!writeBuffer.hasRemaining()) {
				System.out.println("Send order to Server successed!");
			}
		}
	}
}

 

 

 

总结：

1、通过main方法启动Server端 与 客户端

2、 服务器端通过ServerSocketChannel启动监听并且将读事件注册到多路复用器

 

3、客户端通过SocketChannel与服务端建立连接并且将读事件注册到多路复用器

 

4、在接收到消息后 将消息内容读取到缓冲区，然后在进行解码

 

5、发送消息时，将消息读到缓冲区然后在调用write方法将消息发送出去

 

三个重要组件：Buffer、Selector、Channel