使用Java的nio实现高效能的网络通信

在使用Java的API构建网络通信的应用时(尤其是基于移动的网络)，有两种技术可供选择。第一种为直接使用Socket/Server Socket+输入输出流来构建，另外一种使用ServerSocketChannel+Selector。前者为阻塞式通信，需要服务端启动很多线程监听每一个客户端的Socket，通信效率低，且服务端容易受客户端的影响。比如当某个客户端当掉之后，服务的向这个客户端写数据的线程就会一直阻塞，直到写数据超时。所以我们在开发网络通信的应用时，首选后者。如下的代码实现了网络聊天/的功能，在加上语音或视频的录制功能又可实现网络的音频会议或视屏会议。网上和书上有很多类似的例子，但大多没有经过很好的测试，比如有一个客户端突然退出或者当掉，会影响服务端的运行。而我们的实际情况是要求服务端不应该受客户端的影响。

 

代码如下：

 

服务端：

package com.nio;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.Channel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

/**
 * 非阻塞通信的服务端
 * @author Alex
 *
 */
public class NioServer {
	private class NSServerThread extends Thread {
		private String ip;
		private int port;

		public NSServerThread(String ip, int port) {
			this.ip = ip;
			this.port = port;
		}

		public void run() {
			try {
				init(this.ip, this.port);
			} catch (UnknownHostException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public void init(String ip, int port) throws IOException { // 用于检测所有Channel状态的Selector
		System.out.println("打开NioPTTServer 端口为:" + port);
		Selector selector = null;
		// 定义实现编码、解码的字符集对象
		selector = Selector.open();
		// 通过open方法来打开一个未绑定的ServerSocketChannel实例
		ServerSocketChannel server = ServerSocketChannel.open();
		// InetSocketAddress isa = new InetSocketAddress(ip, port);
		InetSocketAddress isa = new InetSocketAddress(ip, port);
		// 将该ServerSocketChannel绑定到指定IP地址
		server.socket().bind(isa);
		// 设置ServerSocket以非阻塞方式工作
		server.configureBlocking(false);
		// 将server注册到指定Selector对象
		server.register(selector, SelectionKey.OP_ACCEPT);
		while (true) {
			int keys = selector.select();
			// System.out.println("keys:"+keys);
			if (keys > 0) {
				// 依次处理selector上的每个已选择的SelectionKey
				// SelectionKey removeReadSk=null;
				try {
					for (SelectionKey sk : selector.selectedKeys()) {
						// 从selector上的已选择Key集中删除正在处理的SelectionKey
						selector.selectedKeys().remove(sk);
						// 如果sk对应的通道包含客户端的连接请求
						if (sk.isAcceptable()) {
							// 调用accept方法接受连接，产生服务器端对应的SocketChannel
							SocketChannel sc = server.accept();
							// 设置采用非阻塞模式
							sc.configureBlocking(false);
							// 将该SocketChannel也注册到selector
							sc.register(selector, SelectionKey.OP_READ);
							// 将sk对应的Channel设置成准备接受其他请求
							sk.interestOps(SelectionKey.OP_ACCEPT);
						}
						// 如果sk对应的通道有数据需要读取
						if (sk.isReadable()) {
							// 获取该SelectionKey对应的Channel，该Channel中有可读的数据
							SocketChannel sc = (SocketChannel) sk.channel();
							// 定义准备执行读取数据的ByteBuffer
							ByteBuffer buff = ByteBuffer.allocate(1024);
							// 开始读取数据
							int len = 0;
							try {
								if ((len = sc.read(buff)) > 0) {
									buff.flip();// 缓存 2指针复位 准备下次读取数据
									System.out.println("读取数据:" + buff.array());
									sk.interestOps(SelectionKey.OP_READ);
								}else{
									System.out.println("没有数据读取,关闭通道");
									sk.cancel();
									if (sk.channel() != null) {
										sk.channel().close();
									}
								}
								// 将sk对应的Channel设置成准备下一次读取
								
							}
							// 如果捕捉到该sk对应的Channel出现了异常，即表明该Channel
							// 对应的Client出现了问题，所以从Selector中取消sk的注册
							catch (IOException ex) {
								// 从Selector中删除指定的SelectionKey
								ex.printStackTrace();
								sk.cancel();
								if (sk.channel() != null) {
									sk.channel().close();
								}
								System.out.println("关闭一个客户端");
							}
							// 如果content的长度大于0，即聊天信息不为空
							if (len > 0) {
								// 遍历该selector里注册的所有SelectKey
								SelectionKey writeKey = null;
								for (SelectionKey key : selector.keys()) {
									// 获取该key对应的Channel

									Channel targetChannel = key.channel();
									// 如果该channel是SocketChannel对象
									if (targetChannel instanceof SocketChannel) {
										// && key != sk) {
										// 将读到的内容写入该Channel中 ,返回到客户端
										if (targetChannel instanceof SocketChannel
												&& key != sk) {
											SocketChannel dest = null;
											try {
												// 将读到的内容写入该Channel中 ,返回到客户端
												dest = (SocketChannel) targetChannel;
												System.out.println("写数据:"
														+ buff.array());
												dest.write(buff);
											} catch (Exception e) {
												// 写异常,关闭通道
												e.printStackTrace();
												if (dest != null) {
													dest.close();
												}
												targetChannel.close();
												writeKey = key;
											}
										}
									}
								}
								if (writeKey != null) {
									writeKey.cancel();
									if (writeKey.channel() != null) {
										writeKey.channel().close();
									}
								}
							}
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
					// selector.close();
					// server.close();
					// System.out.println("发生异常，重新打开端口");
					// init(ip,port);
					// throw new RuntimeException(e);
				}
			}
		}
	}

	public static void main(String[] args) throws IOException {
		// System.out.println(InetAddress.getLocalHost().getHostAddress());
		String host = InetAddress.getLocalHost().getHostAddress();
		int port = 30000;
		new NioServer().new NSServerThread(host, port).start();
		// new NServer().init(InetAddress.getLocalHost().getHostAddress(),
		// 30000);
		System.out.println("Nio服务端启动了,host:" + host);
	}
}

 

 

客户端代码:

package com.nio;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
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.nio.charset.Charset;
import java.util.Scanner;

/**
 * 非阻塞通信的服务端
 * @author Alex
 *
 */
public class NioClient {

	public void init(String ip, int port) throws IOException {
		// 定义检测SocketChannel的Selector对象
		Selector selector = null;
		// 定义处理编码和解码的字符集
		//Charset charset = Charset.forName("UTF-8");
		// 客户端SocketChannel
		SocketChannel sc = null;
		selector = Selector.open();
		InetSocketAddress isa = new InetSocketAddress(ip, port);
		// 调用open静态方法创建连接到指定主机的SocketChannel
		sc = SocketChannel.open(isa);
		// 设置该sc以非阻塞方式工作
		sc.configureBlocking(false);
		// 将SocketChannel对象注册到指定Selector
		sc.register(selector, SelectionKey.OP_READ);
		// 启动读取服务器端数据的线程------------
		new ReadDataThread(selector).start();
		new SendDataThread(sc,selector).start();
	}

	private class SendDataThread extends Thread{
		SocketChannel sc;
		Selector selector;
		public SendDataThread(SocketChannel sc,Selector selector){
			this.sc=sc;
			this.selector=selector;
		}
		public void run(){
			// 创建键盘输入流-----------
			Scanner scan = new Scanner(System.in);
			ByteBuffer byteBuf=ByteBuffer.allocate(152);
			File file=new File("D:/未命名2.jpg");
			try{
				InputStream in=new FileInputStream(file);
				int len=0;
				while (scan.hasNextLine()) {
					// 读取键盘输入
					String line = scan.nextLine();
					
					// 将键盘输入的内容输出到SocketChannel中
					if("send".equals(line.trim())){
						byte[] buf=new byte[152];
						if((len=in.read(buf))>-1){
							byteBuf.put(buf);
							byteBuf.flip();
							sc.write(byteBuf);
							byteBuf.clear();
						}
						byteBuf.put(buf);
						byteBuf.flip();
						sc.write(byteBuf);
						byteBuf.clear();
					}else if("quit".equals(line.trim())){
						sc.close();
						selector.selectedKeys().clear();
						selector.close();
					}
					//sc.write(charset.encode(line));
				}
			}catch(Exception e){
				e.printStackTrace();
			}
			
		}
	}
	//读取服务器数据的----客户端监听线程
	private class ReadDataThread extends Thread {
		Selector selector;

		public ReadDataThread(Selector selector)
				throws IOException {
			this.selector = selector;
		}
		public void run() {
			try {
				ByteBuffer buff = ByteBuffer.allocate(152);
				while (selector.select() > 0) {
					// 遍历每个有可用IO操作Channel对应的SelectionKey
					for (SelectionKey sk : selector.selectedKeys()) {
						// 删除正在处理的SelectionKey
						selector.selectedKeys().remove(sk);
						// 如果该SelectionKey对应的Channel中有可读的数据
						if (sk.isReadable()) {
							// 使用NIO读取Channel中的数据
							SocketChannel sc = (SocketChannel) sk.channel();
							String content = "";
							if (sc.read(buff) > 0) {
								sc.read(buff);
								buff.flip();
								content = new String(buff.array());// charset.decode(buff);
								buff.clear();
							}
							// 打印输出读取的内容
							System.out.println("接收的数据：" + content);
							// 为下一次读取作准备
							sk.interestOps(SelectionKey.OP_READ);
						}
					}
				}
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}
	}

	public static void main(String[] args) throws IOException {
		
		new NioClient().init("192.168.42.128", 30000);
	}
}