JavaNIO处理长连接.doc

JavaNIO处理长连接 最近恰好要写一个处理长连接的服务，接收日志包，然后打包成syslog形式再转发，所以在它的基础上改了一下。主要改了两个类，一个是Server，因为我们只关注read事件，所以write事件我们暂不处理。另外，在处理完ON_READ事件后，不能执行key.cancel()。Java代码 1. packagenioserver; 2. 3. .InetSocketAddress; 4. .ServerSocket; 5. importjava.nio.channels.SelectionKey; 6. importjava.nio.channels.Selector; 7. importjava.nio.channels.ServerSocketChannel; 8. importjava.nio.channels.SocketChannel; 9. importjava.util.Iterator; 10. importjava.util.LinkedList; 11. importjava.util.List; 12. importjava.util.Set; 13. 14. 15. /* 16. *Title:主控服务线程,采用NIO实现的长连接服务器 17. *authorsxl 18. *version1.0 19. */20. 21. publicclassServerimplementsRunnable 22. privatestaticSelectorselector; 23. privateServerSocketChannelsschannel; 24. privateInetSocketAddressaddress; 25. protectedNotifiernotifier; 26. privateintport; 27. 28. /* 29. *创建主控服务线程 30. *paramport服务端口 31. *throwsjava.lang.Exception 32. */33. publicstaticintMAX_THREADS=4; 34. publicServer(intport)throwsException 35. this.port=port; 36. 37. /获取事件触发器 38. notifier=Notifier.getNotifier(); 39. 40. /创建读写线程池 41. for(inti=0;i0) 69. SetselectedKeys=selector.selectedKeys(); 70. Iteratorit=selectedKeys.iterator(); 71. while(it.hasNext() 72. SelectionKeykey=(SelectionKey)it.next(); 73. it.remove(); 74. /处理IO事件 75. if(key.isAcceptable() 76. /Acceptthenewconnection 77. ServerSocketChannelssc=(ServerSocketChannel)key.channel(); 78. notifier.fireOnAccept(); 79. SocketChannelsc=ssc.accept(); 80. sc.configureBlocking(false); 81. /触发接受连接事件 82. Requestrequest=newRequest(sc); 83. notifier.fireOnAccepted(request); 84. /注册读操作,以进行下一步的读操作 85. sc.register(selector,SelectionKey.OP_READ,request); 86. 87. elseif(key.isReadable() 88. RcessRequest(key);/提交读服务线程读取客户端数据 89. 90. 91. 92. 93. catch(Exceptione) 94. continue; 95. 96. 97. 98. package nioserver;import .InetSocketAddress;import .ServerSocket;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.nio.channels.SocketChannel;import java.util.Iterator;import java.util.LinkedList;import java.util.List;import java.util.Set;/* * Title: 主控服务线程,采用NIO实现的长连接服务器 * author sxl * version 1.0 */public class Server implements Runnable private static Selector selector; private ServerSocketChannel sschannel; private InetSocketAddress address; protected Notifier notifier; private int port; /* * 创建主控服务线程 * param port 服务端口 * throws java.lang.Exception */ public static int MAX_THREADS = 4; public Server(int port) throws Exception this.port = port; / 获取事件触发器 notifier = Notifier.getNotifier(); / 创建读写线程池 for (int i = 0; i 0) Set selectedKeys = selector.selectedKeys(); Iterator it = selectedKeys.iterator(); while (it.hasNext() SelectionKey key = (SelectionKey) it.next(); it.remove(); / 处理IO事件 if (key.isAcceptable() / Accept the new connection ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); notifier.fireOnAccept(); SocketChannel sc = ssc.accept(); sc.configureBlocking(false); / 触发接受连接事件 Request request = new Request(sc); notifier.fireOnAccepted(request); / 注册读操作,以进行下一步的读操作 sc.register(selector, SelectionKey.OP_READ, request); else if (key.isReadable() RcessRequest(key); / 提交读服务线程读取客户端数据 catch (Exception e) continue; 另一个改动的类是Reader，改变对-1的处理，这里不是break，而是抛出异常。在读取完buffer的数据后，将数据包传递给另外两个线程进行syslog的发送和入库操作。Java代码 1. packagenioserver; 2. 3. importjava.io.IOException; 4. importjava.nio.ByteBuffer; 5. importjava.nio.channels.SelectionKey; 6. importjava.nio.channels.SocketChannel; 7. importjava.util.LinkedList; 8. importjava.util.List; 9. 10. /* 11. *Title:读线程 12. *Description:该线程用于读取客户端数据 13. *authorsxl 14. *version1.0 15. */16. 17. publicclassReaderextendsThread 18. privatestaticListpool=newLinkedList(); 19. privatestaticNotifiernotifier=Notifier.getNotifier(); 20. 21. publicvoidrun() 22. while(true) 23. try 24. SelectionKeykey; 25. synchronized(pool) 26. while(pool.isEmpty() 27. pool.wait(); 28. 29. key=(SelectionKey)pool.remove(0); 30. 31. /读取数据 32. read(key); 33. 34. catch(Exceptione) 35. continue; 36. 37. 38. 39. 40. /* 41. *读取客户端发出请求数据 42. *paramsc套接通道 43. */44. privatestaticintBUFFER_SIZE=1024; 45. publicstaticbytereadRequest(SocketChannelsc)throwsIOException 46. ByteBufferbuffer=ByteBuffer.allocate(BUFFER_SIZE); 47. intoff=0; 48. intr=0; 49. bytedata=newbyteBUFFER_SIZE*10; 50. while(true) 51. buffer.clear(); 52. r=sc.read(buffer); 53. if(r=0)break; 54. if(r=-1)/如果是流的末尾，则抛出异常 55. thrownewIOException(); 56. if(off+r)data.length)/数组扩容 57. data=grow(data,BUFFER_SIZE*10); 58. 59. bytebuf=buffer.array(); 60. System.arraycopy(buf,0,data,off,r); 61. off+=r; 62. 63. bytereq=newbyteoff; 64. System.arraycopy(data,0,req,0,off); 65. returnreq; 66. 67. 68. /* 69. *处理连接数据读取 70. *paramkeySelectionKey 71. */72. publicvoidread(SelectionKeykey) 73. SocketChannelsc=null; 74. try 75. /读取客户端数据 76. sc=(SocketChannel)key.channel(); 77. byteclientData=readRequest(sc); 78. if(clientData.length0)/有数据才处理 79. Requestrequest=(Request)key.attachment(); 80. request.setDataInput(clientData); 81. /提交到数据库写入线程 82. WcessRequest(request); 83. /提交到Syslog线程,发送syslog 84. ScessRequest(request); 85. 86. 87. catch(Exceptione) 88. if(sc!=null) 89. try 90. sc.socket().close(); 91. sc.close(); 92. catch(IOExceptione1) 93. e1.printStackTrace(); 94. 95. 96. 97. 98. /* 99. *处理客户请求,管理用户的联结池,并唤醒队列中的线程进行处理 100. */101. publicstaticvoidprocessRequest(SelectionKeykey) 102. synchronized(pool) 103. pool.add(pool.size(),key); 104. pool.notifyAll(); 105. 106. 107. 108. /* 109. *数组扩容 110. *paramsrcbyte源数组数据 111. *paramsizeint扩容的增加量 112. *returnbyte扩容后的数组 113. */114. publicstaticbytegrow(bytesrc,intsize) 115. bytetmp=newbytesrc.length+size; 116. System.arraycopy(src,0,tmp,0,src.length); 117. returntmp; 118. 119. package nioserver;import java.io.IOException;import java.nio.ByteBuffer;import java.nio.channels.SelectionKey;import java.nio.channels.SocketChannel;import java.util.LinkedList;import java.util.List;/* * Title: 读线程 * Description: 该线程用于读取客户端数据 * author sxl * version 1.0 */public class Reader extends Thread private static List pool = new LinkedList(); private static Notifier notifier = Notifier.getNotifier(); public void run() while (true) try SelectionKey key; synchronized (pool) while (pool.isEmpty() pool.wait(); key = (SelectionKey) pool.remove(0); / 读取数据 read(key); catch (Exception e) continue; /* * 读取客户端发出请求数据 * param sc 套接通道 */ private static int BUFFER_SIZE = 1024; public static byte readRequest(SocketChannel sc) throws IOException ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE); int off = 0; int r = 0; byte data = new byteBUFFER_SIZE * 10; while ( true ) buffer.clear(); r = sc.read(buffer); if(r = 0) break; if(r = -1)/如果是流的末尾，则抛出异常 throw new IOException(); if (off + r) data.length) /数组扩容 data = grow(data, BUFFER_SIZE * 10); byte buf = buffer.array(); System.arraycopy(buf, 0, data, off, r); off += r; byte req = new byteoff; System.arraycopy(data, 0, req, 0, off); return req; /* * 处理连接数据读取 * param key SelectionKey */ public void read(SelectionKey key) SocketChannel sc = null; try / 读取客户端数据 sc = (SocketChannel) key.channel(); byte clientData = readRequest(sc); if(clientData.length 0)/有数据才处理 Request request = (Request)key.attachment(); request.setDataInput(clientData); / 提交到数据库写入线程 WcessRequest(request); / 提交到Syslog线程,发送syslog ScessReq