nio 例子

作者: 抓兔子的猫 | 来源:发表于2016-09-06 16:09 被阅读160次

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;

/** 
* @author  mengwen  E-mail:  meng_wen@126.com
* @date 创建时间：2016年8月29日 上午11:28:17 
* @version 1.0 
* @parameter  
* @since  
* @return  
*/
public class TCPServer {  
    // 缓冲区大小  
    private static final int BufferSize = 1024;  
    // 超时时间，单位毫秒  
    private static final int TimeOut = 3000;  
    // 本地监听端口  
    private static final int ListenPort = 1978;  
    public static void main(String[] args) throws IOException {  
        // 创建选择器  
        Selector selector = Selector.open();  
        // 打开监听信道  
        ServerSocketChannel listenerChannel = ServerSocketChannel.open();  
        // 与本地端口绑定  
        listenerChannel.socket().bind(new InetSocketAddress(ListenPort));  
        // 设置为非阻塞模式  
        listenerChannel.configureBlocking(false);  
        // 将选择器绑定到监听信道,只有非阻塞信道才可以注册选择器.并在注册过程中指出该信道可以进行Accept操作  
        listenerChannel.register(selector, SelectionKey.OP_ACCEPT);  
        // 创建一个处理协议的实现类,由它来具体操作  
        TCPProtocol protocol = new TCPProtocolImpl(BufferSize);  
  
        // 反复循环,等待IO  
        while (true) {
            // 等待某信道就绪(或超时)  
            if (selector.select(TimeOut) == 0) {// 监听注册通道，当其中有注册的 IO  
                                                // 操作可以进行时，该函数返回，并将对应的  
                                                // SelectionKey 加入 selected-key  
                                                // set  
                System.out.println("独自等待.");
               
                continue;  
            }  
            // 取得迭代器.selectedKeys()中包含了每个准备好某一I/O操作的信道的SelectionKey  
            // Selected-key Iterator 代表了所有通过 select() 方法监测到可以进行 IO 操作的 channel  
            // ，这个集合可以通过 selectedKeys() 拿到  
            Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();  
            while (keyIter.hasNext()) {  
                SelectionKey key = keyIter.next();  
                SelectionKey key1;  
                if (keyIter.hasNext()) {  
                    key1 = keyIter.next();  
                }  
                try {  
                    if (key.isAcceptable()) {  
                        // 有客户端连接请求时  
                        protocol.handleAccept(key);  
                    }  
                    if (key.isReadable()) {// 判断是否有数据发送过来  
                        // 从客户端读取数据  
                        protocol.handleRead(key);  
                    }  
                    if (key.isValid() && key.isWritable()) {// 判断是否有效及可以发送给客户端  
                        // 客户端可写时  
                        protocol.handleWrite(key);  
                    }  
                } catch (IOException ex) {  
                    // 出现IO异常（如客户端断开连接）时移除处理过的键  
                    keyIter.remove();  
                    continue;  
                }  
                // 移除处理过的键  
                keyIter.remove();  
            }  
        }  
    }  
}

package com.dhcc.bugkiller.test.nio;

import java.io.IOException;
import java.nio.channels.SelectionKey;

/** 
* @author  mengwen  E-mail:  meng_wen@126.com
* @date 创建时间：2016年8月29日 上午11:37:50 
* @version 1.0 
* @parameter  
* @since  
* @return  
*/
public interface TCPProtocol{  
       /**  
        * 接收一个SocketChannel的处理  
        * @param key  
        * @throws IOException  
        */  
       void handleAccept(SelectionKey key) throws IOException;  
         
       /**  
        * 从一个SocketChannel读取信息的处理  
        * @param key  
        * @throws IOException  
        */  
       void handleRead(SelectionKey key) throws IOException;  
         
       /**  
        * 向一个SocketChannel写入信息的处理  
        * @param key  
        * @throws IOException  
        */  
       void handleWrite(SelectionKey key) throws IOException;  
     }

package com.dhcc.bugkiller.test.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

import com.sun.star.bridge.oleautomation.Date;


/** 
* @author  mengwen  E-mail:  meng_wen@126.com
* @date 创建时间：2016年8月29日 上午11:41:29 
* @version 1.0 
* @parameter  
* @since  
* @return  
*/
public class TCPProtocolImpl implements TCPProtocol {  
    private int bufferSize;  
  
    public TCPProtocolImpl(int bufferSize) {  
        this.bufferSize = bufferSize;  
    }  
  
    public void handleAccept(SelectionKey key) throws IOException {  
        // 返回创建此键的通道，接受客户端建立连接的请求，并返回 SocketChannel 对象  
        SocketChannel clientChannel = ((ServerSocketChannel) key.channel())  
                .accept();  
        // 非阻塞式  
        clientChannel.configureBlocking(false);  
        // 注册到selector  
        clientChannel.register(key.selector(), SelectionKey.OP_READ,  
                ByteBuffer.allocate(bufferSize));  
    }  
  
    public void handleRead(SelectionKey key) throws IOException {  
        // 获得与客户端通信的信道  
        SocketChannel clientChannel = (SocketChannel) key.channel();  
        // 得到并清空缓冲区  
        ByteBuffer buffer = (ByteBuffer) key.attachment();  
        buffer.clear();  
        // 读取信息获得读取的字节数  
        long bytesRead = clientChannel.read(buffer);  
        if (bytesRead == -1) {  
            // 没有读取到内容的情况  
            clientChannel.close();  
        } else {  
            // 将缓冲区准备为数据传出状态  
            buffer.flip();  
            // 将字节转化为为UTF-16的字符串  
            String receivedString = Charset.forName("UTF-16").newDecoder()  
                    .decode(buffer).toString();  
            // 控制台打印出来  
            System.out.println("接收到来自"  
                    + clientChannel.socket().getRemoteSocketAddress() + "的信息:"  
                    + receivedString);  
            // 准备发送的文本  
            String sendString = "你好,客户端. @" + new Date().toString()  
                    + "，已经收到你的信息" + receivedString;  
            buffer = ByteBuffer.wrap(sendString.getBytes("UTF-16"));  
            clientChannel.write(buffer);  
            // 设置为下一次读取或是写入做准备  
            key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);  
        }  
    }  
  
    public void handleWrite(SelectionKey key) throws IOException {  
        // do nothing  
    }  
}

package com.dhcc.bugkiller.test.nio;

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;
import java.util.Scanner;

/** 
* @author  mengwen  E-mail:  meng_wen@126.com
* @date 创建时间：2016年8月29日 下午1:58:10 
* @version 1.0 
* @parameter  
* @since  
* @return  
*/
public class TCPClient {  
    // 信道选择器  
    private Selector selector;  
  
    // 与服务器通信的信道  
    SocketChannel socketChannel;  
  
    // 要连接的服务器Ip地址  
    private String hostIp;  
  
    // 要连接的远程服务器在监听的端口  
    private int hostListenningPort;  
  
    public TCPClient(String HostIp, int HostListenningPort) throws IOException {  
        this.hostIp = HostIp;  
        this.hostListenningPort = HostListenningPort;  
  
        initialize();  
    }  
    /**  
     * 初始化  
     *   
     * @throws IOException  
     */  
    private void initialize() throws IOException {  
        // 打开监听信道并设置为非阻塞模式  
        socketChannel = SocketChannel.open(new InetSocketAddress(hostIp,  
                hostListenningPort));  
        socketChannel.configureBlocking(false);  
  
        // 打开并注册选择器到信道  
        selector = Selector.open();  
        socketChannel.register(selector, SelectionKey.OP_READ);  
  
        // 启动读取线程  
        new TCPClientReadThread(selector);  
    }  
    /**  
     * 发送字符串到服务器  
     *   
     * @param message  
     * @throws IOException  
     */  
    public void sendMsg(String message) throws IOException {  
        ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes("UTF-16"));  
        socketChannel.write(writeBuffer);  
    }  
    static TCPClient client;  
    static boolean mFlag = true;  
    public static void main(String[] args) throws IOException {  
        client = new TCPClient("localhost", 1978);  
        new Thread() {  
            @Override  
            public void run() {  
                try {  
                    client.sendMsg("你好!Nio!醉里挑灯看剑,梦回吹角连营");  
                    while (mFlag) {  
                        Scanner scan = new Scanner(System.in);//键盘输入数据  
                        String string = scan.next();  
                        client.sendMsg(string);  
                    }  
                } catch (IOException e) {  
                    mFlag = false;  
                    e.printStackTrace();  
                } finally {  
                    mFlag = false;  
                }  
                super.run();  
            }  
        }.start();  
    }
}

package com.dhcc.bugkiller.test.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

/**
 * @author mengwen E-mail: meng_wen@126.com
 * @date 创建时间：2016年8月29日 下午1:33:59
 * @version 1.0
 * @parameter
 * @since
 * @return
 */
public class TCPClientReadThread implements Runnable {  
    private Selector selector;  
  
    public TCPClientReadThread(Selector selector) {  
        this.selector = selector;  
  
        new Thread(this).start();  
    }  
  
    @SuppressWarnings("static-access")  
    public void run() {  
        try {  
            while (selector.select() > 0) {//select()方法只能使用一次，用了之后就会自动删除,每个连接到服务器的选择器都是独立的  
                // 遍历每个有可用IO操作Channel对应的SelectionKey  
                for (SelectionKey sk : selector.selectedKeys()) {  
                    // 如果该SelectionKey对应的Channel中有可读的数据  
                    if (sk.isReadable()) {  
                        // 使用NIO读取Channel中的数据  
                        SocketChannel sc = (SocketChannel) sk.channel();//获取通道信息  
                        ByteBuffer buffer = ByteBuffer.allocate(1024);//分配缓冲区大小  
                        sc.read(buffer);//读取通道里面的数据放在缓冲区内  
                        buffer.flip();// 调用此方法为一系列通道写入或相对获取 操作做好准备  
                        // 将字节转化为为UTF-16的字符串  
                        String receivedString = Charset.forName("UTF-16")  
                                .newDecoder().decode(buffer).toString();  
                        // 控制台打印出来  
                        System.out.println("接收到来自服务器"  
                                + sc.socket().getRemoteSocketAddress() + "的信息:"  
                                + receivedString);  
                        // 为下一次读取作准备  
                        sk.interestOps(SelectionKey.OP_READ);  
                    }  
                    // 删除正在处理的SelectionKey  
                    selector.selectedKeys().remove(sk);  
                }  
            }  
        } catch (IOException ex) {  
            ex.printStackTrace();  
        }  
    }  
}

补充一点：

NIO中的Channel在通信过程中是可以双向通信的。即可读、可写。往Selector注册OP_WRITE，后续可以直接select()出该状态下的Channel进行write操作。操作channell时，正在进行read操作。如果要开始转成write操作。是需要对channel中的bytebuff进行反转（flip）操作。主要是重置position的位置。避免write操作时，直接在read操作后的position上继续往下操作。所以如果你没有进行OP_WRITE这些状态的设置。也是可以“正常”通信的。只不过。channel通信的数据容易出问题（如一边读一边写），所以通过selector注册操作事件，主要是方便筛选出合适的channel来做适合的操作。

nio 例子

补充一点：

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读