JavaNIO处理长连接

之前在IBM的网站上看到过一篇介绍NIO的文章，收获很大。但文中的代码只适合短连接的情况，长连接时就不适用了。

最近恰好要写一个处理长连接的服务，接收日志包，然后打包成syslog形式再转发，所以在它的基础上改了一下。

主要改了两个类，一个是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;
            }
        }
    }
}

 另一个改动的类是Reader，改变对-1的处理，这里不是break，而是抛出异常。在读取完buffer的数据后，将数据包传递给另外两个线程进行syslog的发送和入库操作。

 

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;
    }
}