首页 > 代码库 > 基于NIO的Client/Server程序实践

基于NIO的Client/Server程序实践

转自    http://blog.csdn.net/zhangzhaokun/article/details/6612833

本意是想看明白Zookeeper内部的代码是怎么玩的,在琢磨一段时间之后,发现还是自己先独立写一个基于NIO的C/S模式程序,看看有哪些细微之处要注意,再来跟进ZK的细节比较靠谱一些,于是乎就自己练手写了如下这段代码 ,权当预热下使用NIO来编写网络程序这一知识点了,在这里记述这段代码的目的无非是加深下自己的印象,并且后续还可以有思索和改进的空间。

基本功能:服务器端不停的向Client发送“how are you?”,客户端不停的接收该消息,接收完消息后则向服务器端发送问候语“Server,how are you?”

(1)服务器端程序

package com.hadoop.nio;import java.io.IOException;import java.io.UnsupportedEncodingException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.ClosedChannelException;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.Set;/** * @author Administrator *  */public class NIOSocketServer extends Thread {    private Selector selector;    private ServerSocketChannel ssc;    /**     * @param args     */    public static void main(String[] args) {        NIOSocketServer server = new NIOSocketServer();        try {            // server.setDaemon(true);            server.initServer();            server.start();        } catch (Exception e) {            e.printStackTrace();            server.stopServer();        }    }    public void run() {        while (true) {            try {                int select = selector.select();                if (select > 0) {                    Set<SelectionKey> keys = selector.selectedKeys();                    Iterator<SelectionKey> iter = keys.iterator();                    while (iter.hasNext()) {                        SelectionKey key = iter.next();                        if (key.isAcceptable()) {                            doAcceptable(key);                        }                        if (key.isWritable()) {                            doWriteMessage(key);                        }                        if (key.isReadable()) {                            doReadMessage(key);                        }                        if (key.isConnectable()) {                            doConnectable(key);                        }                        iter.remove();                    }                }            } catch (Exception e) {                e.printStackTrace();            }        }    }    /**     * 初始化服务器端程序,开始监听端口     *      * @throws IOException     * @throws ClosedChannelException     */    private void initServer() throws IOException, ClosedChannelException {        selector = Selector.open();        ssc = ServerSocketChannel.open();        ssc.configureBlocking(false);        ssc.socket().bind(new InetSocketAddress(2181));        ssc.register(selector, SelectionKey.OP_ACCEPT);    }    /**     * 停止服务器端     */    private void stopServer() {        try {            if (selector != null && selector.isOpen()) {                selector.close();            }            if (ssc != null && ssc.isOpen()) {                ssc.close();            }        } catch (IOException e) {            e.printStackTrace();        }    }    /**     * 对新的客户端连接进行处理     *      * @param key     * @throws IOException     * @throws ClosedChannelException     */    private void doAcceptable(SelectionKey key) throws IOException,            ClosedChannelException {        System.out.println("is acceptable");        ServerSocketChannel tempSsc = (ServerSocketChannel) key.channel();        SocketChannel ss = tempSsc.accept();        ss.configureBlocking(false);        ss.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);    }    /**     * 写消息到客户端     *      * @param key     * @throws IOException     * @throws UnsupportedEncodingException     */    private void doWriteMessage(SelectionKey key) throws IOException,            UnsupportedEncodingException {        System.out.println("is writable");        SocketChannel sc = (SocketChannel) key.channel();        ByteBuffer buffer = ByteBuffer.wrap("how are you?".getBytes("UTF-8"));        while (buffer.hasRemaining()) {            sc.write(buffer);        }        // sk.interestOps(SelectionKey.OP_READ);    }    /**     * 读取客户端传递过来的消息     *      * @param key     * @throws IOException     * @throws UnsupportedEncodingException     */    private void doReadMessage(SelectionKey key) throws IOException,            UnsupportedEncodingException {        System.out.println("is readable");        SocketChannel sc = (SocketChannel) key.channel();        ByteBuffer bb = ByteBuffer.allocate(8);        System.out.println("receive from clint:");        int read = sc.read(bb);        while (read > 0) {            bb.flip();            byte[] barr = new byte[bb.limit()];            bb.get(barr);            System.out.print(new String(barr, "UTF-8"));            bb.clear();            read = sc.read(bb);        }        System.out.println("");        // sk.interestOps(SelectionKey.OP_WRITE);    }    /**     * 已连接     *      * @param key     */    private void doConnectable(SelectionKey key) {        System.out.println("is connectalbe");    }}
(2)客户端程序

/** *  */package com.hadoop.nio;import java.io.IOException;import java.io.UnsupportedEncodingException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.ClosedChannelException;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.SocketChannel;import java.util.Iterator;import java.util.Set;/** * @author Administrator *  */public class NIOSocketClient extends Thread {    private SocketChannel socketChannel;    private Selector selector;    /**     * @param args     */    public static void main(String[] args) {        NIOSocketClient client = new NIOSocketClient();        try {            client.initClient();            client.start();            // client.setDaemon(true);        } catch (Exception e) {            e.printStackTrace();            client.stopServer();        }    }    public void run() {        while (true) {            try {                // 写消息到服务器端                writeMessage();                int select = selector.select();                if (select > 0) {                    Set<SelectionKey> keys = selector.selectedKeys();                    Iterator<SelectionKey> iter = keys.iterator();                    while (iter.hasNext()) {                        SelectionKey sk = iter.next();                        if (sk.isReadable()) {                            readMessage(sk);                        }                        iter.remove();                    }                }            } catch (Exception e) {                e.printStackTrace();            }        }    }    public void readMessage(SelectionKey sk) throws IOException,            UnsupportedEncodingException {        SocketChannel curSc = (SocketChannel) sk.channel();        ByteBuffer buffer = ByteBuffer.allocate(8);        while (curSc.read(buffer) > 0) {            buffer.flip();            System.out.println("Receive from server:"                    + new String(buffer.array(), "UTF-8"));            buffer.clear();        }    }    public void writeMessage() throws IOException {        try {            String ss = "Server,how are you?";            ByteBuffer buffer = ByteBuffer.wrap(ss.getBytes("UTF-8"));            while (buffer.hasRemaining()) {                System.out.println("buffer.hasRemaining() is true.");                socketChannel.write(buffer);            }        } catch (IOException e) {            if (socketChannel.isOpen()) {                socketChannel.close();            }            e.printStackTrace();        }    }    public void initClient() throws IOException, ClosedChannelException {        InetSocketAddress addr = new InetSocketAddress(2181);        socketChannel = SocketChannel.open();        selector = Selector.open();        socketChannel.configureBlocking(false);        socketChannel.register(selector, SelectionKey.OP_READ);        // 连接到server        socketChannel.connect(addr);        while (!socketChannel.finishConnect()) {            System.out.println("check finish connection");        }    }    /**     * 停止客户端     */    private void stopServer() {        try {            if (selector != null && selector.isOpen()) {                selector.close();            }            if (socketChannel != null && socketChannel.isOpen()) {                socketChannel.close();            }        } catch (IOException e) {            e.printStackTrace();        }    }}

(3)需要改进的点

1、客户端与服务器端的IO异常等没有处理,程序逻辑上不健壮

2、对于序列化与反序列化后传递的消息,发送端把消息发送完毕,接收端把消息接收完毕如何界定?这里的完毕,如果是不加分析的接收和发送完成那是比较简单的,复杂之处在于如何去判定大数据量的情况下,按照对应的协议把它解析出来?需要一个可扩展的数据协议,后续我会发明一个轮子来解决如下几个问题:A,按照制定分隔符号来解析数据 B:按照【消息长度】【消息内容】这种变长消息来解析数据 C:Java对象的序列化与反序列化

3、监听和处理OP_ACCEPT、OP_READ、OP_WRITE事件的线程数量如何分配才能完成高效的协同,可以根据业务需要调配比率,实现最佳性能?

4、监听OP_WRITE操作比较麻烦,只要写缓冲区没有满则一直可以写,需要频繁的删除和注册OP_WRITE事件,极为影响性能,如何优化?

5、在多个连接同时连接到服务端的时候,一个Selector是否够用,一个Selector能够管理多少Channel呢?另外从Channel中读取消息、向Channel中写消息是否就一定要用SubReactor来处理吗?用WorkThread来处理有什么不好?

6、TCP协议麻烦之处在于接收/发送数据并不能保证有序,在客户端同步等待服务端响应的时候,如何将服务端的异步处理转化为对应于客户端的同步处理?当然客户端也是可以同步或者异步的来访问服务端的。

7、如何避免业务程序的数据与NIO的缓冲区之间相互拷贝,降低性能?怎样才能把一个ByteBuffer的内容从签到后都用一个呢,避免Copy来降低性能。

这些问题的提出,我结合不少网上的资料,也参照了一下Mina、Netty的原理介绍之流的文章,但是并没有去阅读这两个框架的源代码,我寻思着,把这些问题先从理论上解决掉,再自行用代码去实践,把上述的例子给丰满起来,基本解决这些问题之后,再去研读和比较Mina、Netty的具体实现,这样一趟下来就能对Java+NIO+Channel的编程有一个整体的认识了。好几年来,我都想做这个事情,可是一直由于各种原因停滞了,这次务必要实践起来,在这个博文里面提出来这个观点,就在于我想以此宣示来告诉自己务必要一直走到底。

至此,我认为学习一个框架或者是一个知识点的最好办法就是去实践,去以自己的想法发明一个轮子,再模拟场景进行测试,之后再看看开源产品中那些大师是如何实现的,两项对比之下,自己的不足之处就找出来了,也知道该怎么改进了。



基于NIO的Client/Server程序实践