JAVANIOServerSocketChannel(线程池版)
服务器端:
创新互联专注为客户提供全方位的互联网综合服务,包含不限于网站设计、成都做网站、兴山网络推广、小程序定制开发、兴山网络营销、兴山企业策划、兴山品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等,从售前售中售后,我们都将竭诚为您服务,您的肯定,是我们最大的嘉奖;创新互联为所有大学生创业者提供兴山建站搭建服务,24小时服务热线:028-86922220,官方网址:www.cdcxhl.com
import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectableChannel; 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; public class ServerSocketThreadPool{ private static final int MAX_THREAD = Runtime.getRuntime().availableProcessors(); private ThreadPool pool = new ThreadPool(MAX_THREAD); private static int PORT_NUMBER = 1234; public static void main(String[] args) throws Exception { new ServerSocketThreadPool().go(); } public void go() throws Exception { int port = PORT_NUMBER; System.out.println("Listenning on port:" + port); // 创建通道 ServerSocketChannel ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); // 绑定监听端口 serverSocketChannel.socket().bind(new InetSocketAddress(port)); // 设置为非阻塞方式 serverSocketChannel.configureBlocking(false); // 创建选择器 Selector selector = Selector.open(); // 通道注册到选择器 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); while (true) { // 一直阻塞,直到有数据请求 int n = selector.select(); if (n == 0) { continue; } Iteratorit = selector.selectedKeys().iterator(); while (it.hasNext()) { SelectionKey key = it.next(); if (key.isAcceptable()) { ServerSocketChannel server = (ServerSocketChannel) key.channel(); SocketChannel socket = server.accept(); registerChannel(selector,socket, SelectionKey.OP_READ); sayHello(socket); } if (key.isReadable()) { readDataFromSocket(key); } it.remove(); } } } public void registerChannel(Selector selector,SelectableChannel channel,int ops)throws Exception{ if(channel==null){ return; } channel.configureBlocking(false); channel.register(selector, ops); } public void sayHello(SocketChannel socket) throws Exception{ ByteBuffer buffer=ByteBuffer.allocate(1024); buffer.clear(); buffer.put("hello client".getBytes()); buffer.flip(); socket.write(buffer); } public void readDataFromSocket(SelectionKey key) throws Exception { WorkThread thread=pool.getWork(); if(thread==null){ return; } thread.serviceChannel(key); } private class ThreadPool { List idle=new LinkedList(); public ThreadPool(int poolSize) { for(int i=0;i 0){ thread=(WorkThread) idle.remove(0); } } return thread; } public void returnWorker(WorkThread workThread) { synchronized (idle) { idle.add(workThread); } } } private class WorkThread extends Thread { private ByteBuffer buffer = ByteBuffer.allocate(1024); private ThreadPool pool; private SelectionKey key; public WorkThread(ThreadPool pool) { this.pool = pool; } public synchronized void run() { System.out.println(this.getName() + " is ready"); while (true) { try { this.wait(); } catch (InterruptedException e) { e.printStackTrace(); this.interrupt(); } if (key == null) { continue; } System.out.println(this.getName() + " has been awaken"); try{ drainChannel(key); }catch(Exception e){ System.out.println("caught '"+e+"' closing channel"); try{ key.channel().close(); }catch(IOException ioe){ ioe.printStackTrace(); } key.selector().wakeup(); } key=null; this.pool.returnWorker(this); } } synchronized void serviceChannel(SelectionKey key){ this.key=key; key.interestOps(key.interestOps()&(~SelectionKey.OP_READ)); this.notify(); } void drainChannel(SelectionKey key)throws Exception{ SocketChannel channel=(SocketChannel) key.channel(); buffer.clear(); int count; while((count=channel.read(buffer))>0){ buffer.flip(); /*while(buffer.hasRemaining()){ channel.write(buffer); }*/ byte[] bytes; bytes=new byte[count]; buffer.get(bytes); System.out.println(new String(bytes)); buffer.clear(); } if(count<0){ channel.close(); return; } key.interestOps(key.interestOps()|SelectionKey.OP_READ); key.selector().wakeup(); } } }
客户端:
import java.io.ByteArrayOutputStream; 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; public class ClientTest { private static SocketChannel socketChannel; private static Selector selector; public ClientTest()throws Exception { socketChannel=SocketChannel.open(); socketChannel.connect(new InetSocketAddress("127.0.0.1", 1234)); socketChannel.configureBlocking(false); selector=Selector.open(); socketChannel.register(selector, SelectionKey.OP_READ); } public static void main(String[] args) throws Exception{ ClientTest test=new ClientTest(); ByteBuffer buffer=ByteBuffer.allocate(1024); buffer.put("hello server".getBytes()); buffer.flip(); while(buffer.hasRemaining()){ test.socketChannel.write(buffer); } buffer.clear(); socketChannel.socket().shutdownOutput(); String response=receiveData(test.socketChannel); System.out.println(response); } public static String receiveData(SocketChannel socketChannel2) throws IOException { ByteArrayOutputStream baos = new ByteArrayOutputStream(); String response = ""; try { ByteBuffer buffer = ByteBuffer.allocate(1024); byte[] bytes; int count = 0; while ((count = socketChannel2.read(buffer)) >= 0) { buffer.flip(); bytes = new byte[count]; buffer.get(bytes); baos.write(bytes); buffer.clear(); } bytes = baos.toByteArray(); response = new String(bytes); } finally { try { baos.close(); } catch (Exception ex) { } } return response; } }
更详细关于ServerSocketChannel的线程池的理解,请参考《JAVA NIO》这本书的第四章选择器。
文章题目:JAVANIOServerSocketChannel(线程池版)
文章出自:http://scyanting.com/article/ihciss.html