JavaNIO処理長接続

前にIBMのウェブサイトでNIOの記事を紹介するを見たことがありますが、大きな収穫がありました.ただし、文のコードは短い接続の場合にのみ適用され、長い接続の場合は適用されません.
最近はちょうど長い接続を処理するサービスを書いて、ログパケットを受信してsyslog形式にパッケージして転送するので、その上で変更しました.
主に2つのクラスを変更しました.1つはServerです.readイベントだけに注目しているので、writeイベントは処理しません.また、処理完了でON_READイベント後、keyを実行することができない.cancel().

package nioserver;

import java.net.InetSocketAddress;
import java.net.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;


/**
 * <p>Title:       ,  NIO         </p>
 * @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 < MAX_THREADS; i++) {
            Thread r = new Reader();
            Thread w = new Writer();
            Thread sys = new Syslog();
            r.start();
            w.start();
            sys.start();
        }

        //          
        selector = Selector.open();
        sschannel = ServerSocketChannel.open();
        sschannel.configureBlocking(false);
        address = new InetSocketAddress(port);
        ServerSocket ss = sschannel.socket();
        ss.bind(address);
        sschannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    public void run() {
        System.out.println("Server started ...");
        System.out.println("Server listening on port: " + port);
        //   
        while (true) {
            try {
                int num = 0;
                num = selector.select();
                if (num > 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()) {
                           Reader.processRequest(key);  //               
                       }
                    }
                }
            }
            catch (Exception e) {
                continue;
            }
        }
    }
}

もう1つの変更のクラスはReaderであり,1に対する処理を変更し,ここではbreakではなく異常を放出する.bufferのデータを読み出した後、syslogの送信と入庫操作のために他の2つのスレッドにパケットを渡します.
 

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;

/**
 * <p>Title:    </p>
 * <p>Description:             </p>
 * @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 byte[BUFFER_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 byte[off];
        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);
                //           
                Writer.processRequest(request);
                //    Syslog  ,  syslog
                Syslog.processRequest(request);
            }
        }
        catch (Exception e) {
        	if(sc != null)
				try {
					sc.socket().close();
					sc.close();
				} catch (IOException e1) {
					e1.printStackTrace();
				}
        }
    }

    /**
     *       ,        ,             
     */
    public static void processRequest(SelectionKey key) {
        synchronized (pool) {
            pool.add(pool.size(), key);
            pool.notifyAll();
        }
    }

    /**
     *     
     * @param src byte[]      
     * @param size int       
     * @return byte[]       
     */
    public static byte[] grow(byte[] src, int size) {
        byte[] tmp = new byte[src.length + size];
        System.arraycopy(src, 0, tmp, 0, src.length);
        return tmp;
    }
}