当前位置:首页软件开发Java → 底层读写-NIO实战

底层读写-NIO实战

时间:2019-04-01 14:20:43来源:互联网我要评论(0)

     最近参与实现底层高并发处理的项目,接收socket客户端的连接请求,然后调用主机服务进行通讯。系统需要满足100TPS(吞吐量(tps)=活动的用户数/响应时间)的性能要求。
     通过查阅相关的文献及网上的资料,最终采用java NIO技术方案来实现。系统上线后各方面性能表现不错,也给我们技术团队更多的信心。以下分享一下Java NIO这块的程序,希望对各位有帮助。
Java NIO:提供多路(non-bloking) 非阻塞式的高伸缩性网络I/O 。
     Java NIO非堵塞技术实际是采取Reactor模式,或者说是Observer模式为我们监察I/O端口,如果有请求过来,会自动通知监听服务,不需要开启多个线程一直监听请求,从外界看,实现了流畅的I/O读写,不堵塞了。
     机制的实现是通过它的Selector当发现某个channel有数据时,会通过SelectorKey来告知我们,从而实现事件和handler的绑定。
     以下是代码部分,出于项目隐私,相关的业务处理已注释掉,但不影响运行。
服务器端:

Java代码  收藏代码

  1. import java.io.IOException;   import java.net.InetSocketAddress;   import java.net.ServerSocket;   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.nio.charset.Charset;   import java.util.Set;               /**   * @author think   *   */   public class NIOServer implements Runnable{       private int port = 7070;       // 解码buffer       private Charset cs = Charset.forName("gbk");       private static Selector selector;       public NIOServer() {       }          /**       * 启动服务器端,配置为非阻塞,绑定端口,注册accept事件ACCEPT事件:当服务端收到客户端连接请求时,触发该事件       * @throws IOException       */       private void initchannel(){           try{               //初始化SOCKET通道               ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();               serverSocketChannel.configureBlocking(false);               ServerSocket serverSocket = serverSocketChannel.socket();               serverSocket.bind(new InetSocketAddress(port));               selector = Selector.open();               serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);               System.out.println("Init NIOServer listening port:" + port);           } catch (Exception e) {               System.out.println("Init NIOServer failed! error:"+e.getMessage());           }                  }          /**       * 服务器端轮询监听,select方法会一直阻塞直到有相关事件发生或超时       */       private void listen() {           //初始化SOCKET通道           initchannel();           while (true) {               try {                   selector.select();// 返回值为本次触发的事件数                   Set<SelectionKey> selectionKeys = selector.selectedKeys();                   for (SelectionKey key : selectionKeys) {                       //处理客户端请求                       handle(key);                   }                   selectionKeys.clear();// 清除处理过的事件               } catch (Exception e) {                   System.out.println("Exit listen port:"+this.port+" error:"+e.getMessage());               }              }       }          /**       * 处理不同的事件       */       private void handle(SelectionKey selectionKey) throws IOException {           ServerSocketChannel server = null;           SocketChannel client = null;           if (selectionKey.isAcceptable()) {               /*               * 客户端请求连接事件 serversocket为该客户端建立socket连接,将此socket注册READ事件,监听客户端输入               * READ事件:当客户端发来数据,并已被服务器控制线程正确读取时,触发该事件               */               server = (ServerSocketChannel) selectionKey.channel();               client = server.accept();               //System.out.println("Client IP:"+client.getRemoteAddress().toString());               client.configureBlocking(false);               client.register(selector, SelectionKey.OP_READ);           } else if (selectionKey.isReadable()) {               ByteBuffer sBuffer = ByteBuffer.allocate(1024);               /*               * READ事件,收到客户端发送数据,读取数据后继续注册监听客户端               */               client = (SocketChannel) selectionKey.channel();               sBuffer.clear();                  int n = -1;               try {                   n = client.read(sBuffer);                   //log.info("输入字节数:" + n);               } catch (Exception e) {                   n=-1;                   client.close();                   selectionKey.cancel();               }               if (n > 0) {                   sBuffer.flip();                   String receiveText = String.valueOf(cs.decode(sBuffer).array());                   //实际业务处理部分忽略                   channelWriteBytes(client,"Server:" +receiveText);               }           }       }          @Override       public void run() {           //线程开启SOCKET监听           listen();       }       /**       * 结果回写       */       public void channelWriteBytes(SocketChannel socketChannel,String result){           ByteBuffer byBuffer = ByteBuffer.allocate(10240);           byBuffer.put((result).getBytes(Charset.forName("gbk")));           byBuffer.flip();           // 输出到通道           try {               socketChannel.write(byBuffer);           } catch (IOException e) {               System.out.println("NIOServer write back failed:"+e.getMessage());           }       }              public static void main(String[] args) throws IOException {           NIOServer server = new NIOServer();           server.listen();       }      }  

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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.nio.charset.Charset;
import java.util.Set;




/**
 * @author think
 *
 */
public class NIOServer implements Runnable{
	private int port = 7070;
	// 解码buffer
	private Charset cs = Charset.forName("gbk");
	private static Selector selector;
	public NIOServer() {
	}

	/**
	 * 启动服务器端,配置为非阻塞,绑定端口,注册accept事件ACCEPT事件:当服务端收到客户端连接请求时,触发该事件
	 * @throws IOException
	 */
	private void initchannel(){
		try{
			//初始化SOCKET通道
			ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
			serverSocketChannel.configureBlocking(false);
			ServerSocket serverSocket = serverSocketChannel.socket();
			serverSocket.bind(new InetSocketAddress(port));
			selector = Selector.open();
			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
			System.out.println("Init NIOServer listening port:" + port);
		} catch (Exception e) {
			System.out.println("Init NIOServer failed! error:"+e.getMessage());
		}
		
	}

	/**
	 * 服务器端轮询监听,select方法会一直阻塞直到有相关事件发生或超时
	 */
	private void listen() {
		//初始化SOCKET通道
		initchannel();
		while (true) {
			try {
				selector.select();// 返回值为本次触发的事件数
				Set<SelectionKey> selectionKeys = selector.selectedKeys();
				for (SelectionKey key : selectionKeys) {
					//处理客户端请求
					handle(key);
				}
				selectionKeys.clear();// 清除处理过的事件
			} catch (Exception e) {
				System.out.println("Exit listen port:"+this.port+" error:"+e.getMessage());
			}

		}
	}

	/**
	 * 处理不同的事件
	 */
	private void handle(SelectionKey selectionKey) throws IOException {
		ServerSocketChannel server = null;
		SocketChannel client = null;
		if (selectionKey.isAcceptable()) {
			/*
			 * 客户端请求连接事件 serversocket为该客户端建立socket连接,将此socket注册READ事件,监听客户端输入
			 * READ事件:当客户端发来数据,并已被服务器控制线程正确读取时,触发该事件
			 */
			server = (ServerSocketChannel) selectionKey.channel();
			client = server.accept();
			//System.out.println("Client IP:"+client.getRemoteAddress().toString());
			client.configureBlocking(false);
			client.register(selector, SelectionKey.OP_READ);
		} else if (selectionKey.isReadable()) {
			ByteBuffer sBuffer = ByteBuffer.allocate(1024);
			/*
			 * READ事件,收到客户端发送数据,读取数据后继续注册监听客户端
			 */
			client = (SocketChannel) selectionKey.channel();
			sBuffer.clear();

			int n = -1;
			try {
				n = client.read(sBuffer);
				//log.info("输入字节数:" + n);
			} catch (Exception e) {
				n=-1;
				client.close();
				selectionKey.cancel();
			}
			if (n > 0) {
				sBuffer.flip();
				String receiveText = String.valueOf(cs.decode(sBuffer).array());
				//实际业务处理部分忽略
				channelWriteBytes(client,"Server:" +receiveText);
			}
		}
	}

	@Override
	public void run() {
		//线程开启SOCKET监听
		listen();
	}
	/**
     * 结果回写
     */
    public void channelWriteBytes(SocketChannel socketChannel,String result){
    	ByteBuffer byBuffer = ByteBuffer.allocate(10240);
    	byBuffer.put((result).getBytes(Charset.forName("gbk")));
		byBuffer.flip();
		// 输出到通道
		try {
			socketChannel.write(byBuffer);
		} catch (IOException e) {
			System.out.println("NIOServer write back failed:"+e.getMessage());
		}
    }
    
	public static void main(String[] args) throws IOException {
		NIOServer server = new NIOServer();
		server.listen();
	}

}


客户端进行调用即可:

Java代码  收藏代码

  1. import java.net.Socket;      /**   * @author think   */      public class Client{        public static void main(String[] args) {               try {                   Socket socket = new Socket("127.0.0.1", 7070);                   String res = null;                   try {                       String request = "NIO Test...";                       byte[] bts = request.getBytes();                       socket.getOutputStream().write(bts);                       socket.getOutputStream().flush();                                  byte[] resb = new byte[60000];                       socket.getInputStream().read(resb);                       res = new String(resb);                       System.out.println(res);                   } catch (Exception e) {                       e.printStackTrace();                   }               } catch (Exception e) {                   // TODO Auto-generated catch block                   e.printStackTrace();               }           }   }  

import java.net.Socket;

/**
 * @author think
 */

public class Client{
	 public static void main(String[] args) {
			try {
				Socket socket = new Socket("127.0.0.1", 7070);
				String res = null;
				try {
					String request = "NIO Test...";
					byte[] bts = request.getBytes();
					socket.getOutputStream().write(bts);
					socket.getOutputStream().flush();			
					byte[] resb = new byte[60000];
					socket.getInputStream().read(resb);
					res = new String(resb);
					System.out.println(res);
				} catch (Exception e) {
					e.printStackTrace();
				}
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
}


      其实实现系统优异的高并发性能,还涉及到更多的方面去学习。例如并发锁的控制,及线程安全的问题。

相关文章

网友评论

热门评论

最新评论

发表评论 查看所有评论()

昵称:
表情: 高兴 可 汗 我不要 害羞 好 下下下 送花 屎 亲亲
字数: 0/500 (您的评论需要经过审核才能显示)

关于万荚 | 联系方式 | 发展历程 | 版权声明 | 帮助(?) | 网站地图 | 友情链接

Copyright 2005-2019 16WJ.COM 〖万荚网〗 版权所有 桂ICP备18000060号 |

声明: 本站所有文章来自互联网 如有异议 请与本站联系