国内最全IT社区平台 联系我们 | 收藏本站
华晨云阿里云优惠2
您当前位置:首页 > php框架 > 框架设计 > Java之NIO(二)selector socketChannel

Java之NIO(二)selector socketChannel

来源:程序员人生   发布时间:2015-06-19 08:25:49 阅读次数:5786次

上篇文章对NIO进行了简介,对Channel和Buffer接口的使用进行了说明,并举了1个简单的例子来讲明其使用方法。

本篇则重点说明selector,Selector(选择器)是Java NIO中能够检测1到多个NIO通道,并能够知晓通道是不是为诸如读写事件做好准备的组件。这样,1个单独的线程可以管理多个channel,从而管理多个网络连接。

与selector联系紧密的是ServerSocketChannel和SocketChannel,他们的使用与上篇文章描写的FileChannel的使用方法类似,然后与ServerSocket和Socket也有1些联系。

本篇首先简单的进selector进行说明,然后1个简单的示例程序,来演示即时通讯。

Selector

使用传统IO进行网络编程,以下图所示:


每个到服务真个连接,都需要1个单独的线程(或线程池)来处理其对应的socket,当连接数多的时候,对服务真个压力极大。并使用socket的getInputStream。Read方法来不断的轮训每一个socket,效力可想而知。

而selector则可以在同1个线程中监听多个channel的状态,当某个channel有selector感兴趣的事情发现,selector则被激活。即不会主动去轮询。以下图所示:

 

Selector使用以下示意:

public static void main(String[] args) throws IOException { Selector selector = Selector.open();//声明selector ServerSocketChannel sc = ServerSocketChannel.open(); sc.configureBlocking(false);//必须设置为异步 sc.socket().bind(new InetSocketAddress(8081));//绑定端口 //把channel 注册到 selector上 sc.register(selector, SelectionKey.OP_ACCEPT|SelectionKey.OP_CONNECT|SelectionKey.OP_READ|SelectionKey.OP_WRITE); while(true){ selector.select();//阻塞,直到注册的channel上某个感兴趣的事情产生 Set<SelectionKey> selectedKeys = selector.selectedKeys(); Iterator<SelectionKey> keyIterator = selectedKeys.iterator(); while(keyIterator.hasNext()) { SelectionKey key = keyIterator.next(); if(key.isAcceptable()) { // a connection was accepted by a ServerSocketChannel. } else if (key.isConnectable()) { // a connection was established with a remote server. } else if (key.isReadable()) { // a channel is ready for reading } else if (key.isWritable()) { // a channel is ready for writing } keyIterator.remove(); } } }

 

极简即时通讯

本例子是是1个极其简单的例子,很多地方都不完善,但是例子可以很好的说明selector的使用方法。

本例子包括服务端和客户端两个部份,其中服务端采取两个selector,用来建立连接和数据的读写。两个selector在两个线程中。

服务端

/** * 简单的即时通讯服务端,采取建立连接 selector和数据 selector分离。很不完善 * */ public class ServerSocketChannelTest { private static final int SERVER_PORT = 8081; private ServerSocketChannel server; private volatile Boolean isStop = false; //负责建立连接的selector private Selector conn_Sel; //负责数据读写的selector private Selector read_Sel; // private ExecutorService sendService = Executors.newFixedThreadPool(3); //锁,用来在建立连接后,唤醒read_Sel时使用的同步 private Object lock = new Object(); //注册的用户 private Map<String, ClientInfo> clents = new HashMap<String, ClientInfo>(); /** * 初始化,绑定端口 */ public void init() throws IOException { //创建ServerSocketChannel server = ServerSocketChannel.open(); //绑定端口 server.socket().bind(new InetSocketAddress(SERVER_PORT)); server.configureBlocking(false); //定义两个selector conn_Sel = Selector.open(); read_Sel = Selector.open(); //把channel注册到selector上,第2个参数为兴趣的事件 server.register(conn_Sel, SelectionKey.OP_ACCEPT); } // 负责建立连接。 private void beginListen() { System.out.println("--------开始监听----------"); while (!isStop) { try { conn_Sel.select(); } catch (IOException e) { e.printStackTrace(); continue; } Iterator<SelectionKey> it = conn_Sel.selectedKeys().iterator(); while (it.hasNext()) { SelectionKey con = it.next(); it.remove(); if (con.isAcceptable()) { try { SocketChannel newConn = ((ServerSocketChannel) con .channel()).accept(); handdleNewInConn(newConn); } catch (IOException e) { e.printStackTrace(); continue; } } else if (con.isReadable()) {//废代码,履行不到。 try { handleData((SocketChannel) con.channel()); } catch (IOException e) { e.printStackTrace(); } } } } } /** * 负责接收数据 */ private void beginReceive(){ System.out.println("---------begin receiver data-------"); while (true) { synchronized (lock) { } try { read_Sel.select(); } catch (IOException e) { e.printStackTrace(); continue; } Iterator<SelectionKey> it = read_Sel.selectedKeys().iterator(); while (it.hasNext()) { SelectionKey con = it.next(); it.remove(); if (con.isReadable()) { try { handleData((SocketChannel) con.channel()); } catch (IOException e) { e.printStackTrace(); } } } } } private void handdleNewInConn(SocketChannel newConn) throws IOException { newConn.configureBlocking(false); //这里必须先唤醒read_Sel,然后加锁,避免读写线程的中select方法再次锁定。 synchronized (lock) { read_Sel.wakeup(); newConn.register(read_Sel, SelectionKey.OP_READ); } //newConn.register(conn_Sel, SelectionKey.OP_READ); } private void handleData(final SocketChannel data) throws IOException { ByteBuffer buffer = ByteBuffer.allocate(512); try { int size= data.read(buffer); if (size==⑴) { System.out.println("-------连接断开-----"); //这里暂时不处理,这里可以移除已注册的客户端 } } catch (IOException e) { e.printStackTrace(); return; } buffer.flip(); byte[] msgByte = new byte[buffer.limit()]; buffer.get(msgByte); Message msg = Message.getMsg(new String(msgByte)); //这里读完数据其实已可以另开线程了下1步的处理,理想情况下,根据不同的消息类型,建立不同的队列,把待发送的消息放进队列 //固然也能够持久化。如果在数据没有读取前,另开线程的话,读写线程中 read_Sel.select(),会立刻返回。可以把 if (msg.getType().equals("0")) {// 注册 ClientInfo info = new ClientInfo(msg.getFrom(), data); clents.put(info.getClentID(), info); System.out.println(msg.getFrom() + "注册成功"); } else {// 转发 System.out.println("收到"+msg.getFrom()+"发给"+msg.getTo()+"的消息"); ClientInfo to = clents.get(msg.getTo()); buffer.rewind(); if (to != null) { SocketChannel sendChannel = to.getChannel(); try { while (buffer.hasRemaining()) { sendChannel.write(buffer); } } catch (Exception e) { } finally { buffer.clear(); } } } } public static void main(String[] args) throws IOException { final ServerSocketChannelTest a = new ServerSocketChannelTest(); a.init(); new Thread("receive..."){ public void run() { a.beginReceive(); }; }.start(); a.beginListen(); } }


客户端


/** * new 次对象,然后调用start方法,其中self 是自己id * * to 是接收人id * */ public class Client { /** * 自己的ID */ private String self; /** * 接收人ID */ private String to; //通道管理器 private Selector selector; private ByteBuffer writeBuffer = ByteBuffer.allocate(512); private SocketChannel channel; private Object lock = new Object(); private volatile boolean isInit = false; public Client(String self, String to) { super(); this.self = self; this.to = to; } /** * 取得1个Socket通道,并对该通道做1些初始化的工作 * @param ip 连接的服务器的ip * @param port 连接的服务器的端口号 * @throws IOException */ public void initClient(String ip,int port) throws IOException { // 取得1个Socket通道 channel = SocketChannel.open(); // 设置通道为非阻塞 channel.configureBlocking(false); // 取得1个通道管理器 this.selector = Selector.open(); // 客户端连接服务器,其实方法履行并没有实现连接,需要在listen()方法中调 //用channel.finishConnect();才能完成连接 channel.connect(new InetSocketAddress(ip,port)); //将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_CONNECT事件。 channel.register(selector, SelectionKey.OP_CONNECT); } /** * 采取轮询的方式监听selector上是不是有需要处理的事件,如果有,则进行处理 * @throws IOException */ @SuppressWarnings("unchecked") public void listen() throws IOException { // 轮询访问selector while (true) { synchronized (lock) { } selector.select(); // 取得selector当选中的项的迭代器 Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator(); while (ite.hasNext()) { SelectionKey key = ite.next(); // 删除已选的key,以防重复处理 ite.remove(); // 连接事件产生 if (key.isConnectable()) { SocketChannel channel = (SocketChannel) key .channel(); // 如果正在连接,则完成连接 if(channel.isConnectionPending()){ channel.finishConnect(); } // 设置成非阻塞 channel.configureBlocking(false); //在和服务端连接成功以后,为了可以接收到服务真个信息,需要给通道设置读的权限。 channel.register(this.selector, SelectionKey.OP_READ); isInit = true; // 取得了可读的事件 } else if (key.isReadable()) { read(key); } } } } /** * 处理读取服务端发来的信息的事件 * @param key * @throws IOException */ public void read(SelectionKey key) throws IOException{ SocketChannel data = (SocketChannel) key.channel(); ByteBuffer buffer = ByteBuffer.allocate(512) ; try { data.read(buffer ); } catch (IOException e) { e.printStackTrace(); data.close(); return; } buffer.flip(); byte[] msgByte = new byte[buffer.limit()]; buffer.get(msgByte); Message msg = Message.getMsg(new String(msgByte)); System.out.println("---收到消息--"+msg+" 来自 "+msg.getFrom()); } private void sendMsg(String content){ writeBuffer.put(content.getBytes()); writeBuffer.flip(); try { while (writeBuffer.hasRemaining()) { channel.write(writeBuffer); } } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } writeBuffer.clear(); } /** * 启动客户端测试 * @throws IOException */ public void start() throws IOException { initClient("localhost",8081); new Thread("reading"){ public void run() { try { listen(); } catch (IOException e) { e.printStackTrace(); } }; }.start(); int time3 = 0; while(!isInit&&time3<3){ try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } time3 ++; } System.out.println("--------开始注册------"); Message re = new Message("", self, ""); sendMsg(re.toString()); try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("-----注册成功----"); String content =""; System.out.println("---- 请输入要发送的消息,按回车发送,输入 123 退出----------"); Scanner s = new Scanner(System.in); while (!content.equals("123")&&s.hasNext()) { content = s.next(); Message msg = new Message(content, self, to); msg.setType("1"); sendMsg(msg.toString()); if (content.equals("123")) { break; } System.out.println("---发送成功---"); } channel.close(); } }


客户端测试

public class TestClient1 { public static void main(String[] args) throws IOException { Client c1 =new Client("1", "2"); c1.start(); } } public class TestClient2 { public static void main(String[] args) throws IOException { Client c2 =new Client("2", "1"); c2.start(); } }

结束

本文的例子极其简单,但是都经过测试。在编码的进程中,遇到的问题主要有两点:

1.     channel.register()方法阻塞

2.     使用线程池遇到问题。本文最后在服务真个读写线程中,没有使用线程池,缘由注释说的比较明白,也说明了使用线程池的1种假想。

 

另外在本文编码进程中,遇到了1些问题,去网上寻求答案,遇到了1些不错的文章,本文某些部份由参考。

selector的讲授,官方文档翻译 http://ifeve.com/selectors/

NIO就绪的OP_write http://blog.csdn.net/zhouhl_cn/article/details/6582435

此文不错:http://blog.csdn.net/jjzhk/article/details/39553613

http://www.2cto.com/kf/201312/267592.html

 

另外还有两个反面教材:

http://www.oschina.net/code/snippet_860673_22507毛病很大

http://www.oschina.net/code/snippet_246601_22883代码本身是正确的,但底下的评论人没有好好看书。

 

 

生活不易,码农辛苦
如果您觉得本网站对您的学习有所帮助,可以手机扫描二维码进行捐赠
程序员人生
------分隔线----------------------------
分享到:
------分隔线----------------------------
关闭
程序员人生