Java网络编程由浅入深三 一文了解非阻塞通信

本文详细介绍组成非阻塞通信的几大类：Buffer、Channel、Selector、SelectionKey

非阻塞通信的流程

  1. ServerSocketChannel通过open方法获取ServerSocketChannel，通过ServerSocketChannel设置为非阻塞模式，再通过ServerSocketChannel获取socket，绑定服务进程监听端口。服务启动成功。
  2. 然后就是非阻塞通信的精髓了，Selector通过静态的open()方法获取到Selector，然后ServerSocketChannel注册Selection.OP_ACCEPT事件到Selector上。
  3. Selector就会监控事件发生，Selector通过select()监控已发生的SelectionKey对象的数目，通过selectKeys()方法返回对应的selectionKey对象集合。遍历该集合得到相应的selectionKey对象，通过该对象的channel()方法获取关联的ServerSocketChannel对象， 通过selector()方法就可以获取关联的Selector对象。
  4. 通过上面获取的ServerSocketChannel执行accept()方法获取SocketChannel，再通过SocketChannel设置为非阻塞模式，在将SocketChannel注册到上面创建的Selector上，注册SelectionKey.OP_READ |SelectionKey.OP_WRITE 事件。
  5. Selector将在监控对应上面绑定的事件，监控到对应的事件的话执行读和写的操作。

示例代码：

上面描述了服务端非阻塞方式通信的一个流程，下面通过具体代码实现：

/**
 * 非阻塞模式
 * 
 */
public class EchoServer2 {
    private Selector selector = null;
    private ServerSocketChannel serverSocketChannel = null;
    private int port = 8001;
    private Charset charset = Charset.forName("UTF-8");

    public EchoServer2() throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        //服务器重启的时候，重用端口
        serverSocketChannel.socket().setReuseAddress(true);
        //设置非阻塞模式
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        System.out.println("服务器启动成功");
    }

    /**
     * 服务方法
     */
    public void service() throws IOException {
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (selector.select() > 0) {
            Set readyKes = selector.selectedKeys();
            Iterator it = readyKes.iterator();
            while (it.hasNext()) {
                SelectionKey key = null;
                try {
                    key = (SelectionKey) it.next();
                    it.remove();
                    if (key.isAcceptable()) {
                        System.out.println("连接事件");
                        //连接事件
                        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                        SocketChannel socketChannel = ssc.accept();
                        System.out.println("接收到客户连接，来自：" + socketChannel.socket().getInetAddress() +
                                " : " + socketChannel.socket().getPort());
                        socketChannel.configureBlocking(false);
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        socketChannel.register(selector, SelectionKey.OP_READ |
                                SelectionKey.OP_WRITE, buffer);
                    } else if (key.isReadable()) {
                        //接收数据
                        receive(key);
                    } else if (key.isWritable()) {
                        //发送数据
                        send(key);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    try {
                        if (key != null) {
                            key.cancel();
                            key.channel().close();
                        }
                    }catch (IOException ex){
                        ex.printStackTrace();
                    }
                }
            }
        }
    }

    private void send(SelectionKey key) throws IOException {
        ByteBuffer buffer = (ByteBuffer) key.attachment();

        SocketChannel channel = (SocketChannel) key.channel();
        buffer.flip(); //把极限设置为位置，把位置设置为0
        String data = decode(buffer);
        if (data.indexOf("\r\n") == -1) {
            return;
        }
        String outputData = data.substring(0, data.indexOf("\n") + 1);
        System.out.println("请求数据：" + outputData);

        ByteBuffer outputBuffer = encode("echo:" + outputData);
        while (outputBuffer.hasRemaining()) {
            channel.write(outputBuffer);
        }
        ByteBuffer temp = encode(outputData);
        buffer.position(temp.limit());
        buffer.compact();

        if (outputData.equals("bye\r\n")) {
            key.cancel();
            channel.close();
            System.out.println("关闭与客户的连接");
        }
    }

    private String decode(ByteBuffer buffer) {
        CharBuffer charBuffer = charset.decode(buffer);
        return charBuffer.toString();
    }

    private ByteBuffer encode(String s) {
        return charset.encode(s);
    }


    private void receive(SelectionKey key) throws IOException {
        ByteBuffer buffer = (ByteBuffer) key.attachment();

        SocketChannel socketChannel = (SocketChannel) key.channel();
        ByteBuffer readBuff = ByteBuffer.allocate(32);
        socketChannel.read(readBuff);
        readBuff.flip();

        buffer.limit(buffer.capacity());
        buffer.put(readBuff);
    }

    public static void main(String[] args) throws IOException {
        new EchoServer2().service();
    }
}

/**
 * 创建非阻塞客户端
 * 
 */
public class EchoClient2 {

    private SocketChannel socketChannel;
    private int port = 8001;
    private Selector selector;
    private ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
    private ByteBuffer receiveBuffer = ByteBuffer.allocate(1024);
    private Charset charset = Charset.forName("UTF-8");

    public EchoClient2() throws IOException {
        socketChannel = SocketChannel.open();
        InetSocketAddress inetSocketAddress = new InetSocketAddress(InetAddress.getLocalHost(), port);
        socketChannel.connect(inetSocketAddress);//
        socketChannel.configureBlocking(false);//设置为非阻塞模式
        System.out.println("与服务器连接成功");
        selector = Selector.open();
    }

    public static void main(String[] args) throws IOException {
        final EchoClient2 client = new EchoClient2();
        Thread receiver = new Thread(new Runnable() {
            @Override
            public void run() {
                client.receiveFromUser();
            }
        });
        receiver.start();
        client.talk();
    }

    private void receiveFromUser() {
        try {
            System.out.println("请输入数据：");
            BufferedReader localReader = new BufferedReader(new InputStreamReader(System.in));
            String msg = null;
            while ((msg = localReader.readLine()) != null) {
                System.out.println("用户输入的数据：" + msg);
                synchronized (sendBuffer) {
                    sendBuffer.put(encode(msg + "\r\n"));
                }
                if (msg.equalsIgnoreCase("bye")) {
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private ByteBuffer encode(String s) {
        return charset.encode(s);
    }

    private void talk() throws IOException {
        socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        while (selector.select() > 0) {
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();
            while (it.hasNext()) {
                SelectionKey key = null;
                try {
                    key = it.next();
                    it.remove();
                    if (key.isReadable()) {
                        //System.out.println("读事件");
                        //读事件
                        receive(key);
                    }
                    if (key.isWritable()) {
                       // System.out.println("写事件");
                        //写事件
                        send(key);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    if (key != null) {
                        key.cancel();
                        key.channel().close();
                    }
                }
            }
        }
    }

    private void send(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        synchronized (sendBuffer) {
            sendBuffer.flip();//把极限设为位置，把位置设为零
            channel.write(sendBuffer);
            sendBuffer.compact();//删除已经发送的数据。
        }
    }

    private void receive(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        channel.read(receiveBuffer);
        receiveBuffer.flip();//将limit的值设置为position的值，将position的值设置为0
        String receiveData = decode(receiveBuffer);
        if (receiveData.indexOf("\n") == -1) {
            return;
        }
        String outputData = receiveData.substring(0, receiveData.indexOf("\n") + 1);
        System.out.println("响应数据：" + outputData);

        if (outputData.equalsIgnoreCase("echo:bye\r\n")) {
            key.cancel();
            socketChannel.close();
            ;
            System.out.println("关闭与服务器的连接");
            selector.close();
            System.exit(0);
        }

        ByteBuffer temp = encode(outputData);
        receiveBuffer.position(temp.limit());
        receiveBuffer.compact();//删除已经打印的数据
    }

    private String decode(ByteBuffer receiveBuffer) {
        CharBuffer buffer = charset.decode(receiveBuffer);
        return buffer.toString();
    }
}

实现非阻塞通信的方式

  • 缓冲区
  • 通道
  • Selector

缓冲区

作用：减少物理读写次数，减少内存创建和销毁次数。 缓冲区的属性：capacity(最大容量)、limit(实际容量)、position(当前位置)。<font color='red'><b>PS：</b></font>其他地方是翻译成capacity(容量)、limit(极限)、position位置)，我个人觉得翻译成上面的更好理解，为啥通过下面的方法解析和图解就可明白。当然最好通过英文表达这样最清楚。
三个属性的关系为：capacity≥limit≥position≥0

图解关系如下：


这里写图片描述

缓冲区类结构：
java.nio.ByteBuffer类是一个抽象类，不能被实例化。但是提供了8个具体的实现类，其中最基本的的缓冲区是ByteBuffer，它存放的数据单元是字节。


这里写图片描述

常用方法：

clear()：把limit设置为capacity，再把位置设为0
flip()：把limit设置为position，再把位置设置为0。
rewind()：不改变limit，把位置设为0。
allocate():创建一个缓冲中，方法参数指定缓冲区大小
compact():将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。

测试上述方法：
测试clear()方法

    @Test
    public void testClear() {
        //创建一个10chars大小的缓冲区，默认情况下limit和capacity是相等的
        CharBuffer buffer = CharBuffer.allocate(10);
        System.out.println("创建默认情况");
        printBufferInfo(buffer);
        buffer.limit(8);//修改limit的值
        System.out.println("修改limit后");
        printBufferInfo(buffer);
        // clear()：把limit设置为capacity，再把位置设为0
        buffer.clear();
        System.out.println("执行clear()方法后");
        printBufferInfo(buffer);
    }

执行结果如下：


这里写图片描述

测试flip()方法：

    @Test
    public void testFlip() {
        CharBuffer buffer = CharBuffer.allocate(10);
        System.out.println("创建默认情况");
        printBufferInfo(buffer);
        //put的方法会修改position的值
        buffer.put('H');
        buffer.put('E');
        buffer.put('L');
        buffer.put('L');
        buffer.put('O');
        System.out.println("调用put方法后：");
        printBufferInfo(buffer);
        //flip()：把limit设置为position，再把位置设置为0。
        buffer.flip();
        System.out.println("调用flip方法后：");
        printBufferInfo(buffer);
    }

执行结果如下：


这里写图片描述

测试rewind()方法

        @Test
    public void testRewind() {
        CharBuffer buffer = CharBuffer.allocate(10);
        System.out.println("创建默认情况");
        printBufferInfo(buffer);
        //put的方法会修改position的值
        buffer.put('H');
        buffer.put('E');
        buffer.put('L');
        buffer.put('L');
        buffer.put('O');
        buffer.limit(8);
        System.out.println("调用put、limit方法后：");
        printBufferInfo(buffer);
        //rewind()：不改变limit，把位置设为0。
        buffer.rewind();
        System.out.println("调用rewind方法后：");
        printBufferInfo(buffer);
    }

执行结果如下：


这里写图片描述

测试compact()方法

    @Test
    public void testCompact(){
        CharBuffer buffer = CharBuffer.allocate(10);
        System.out.println("创建默认情况");
        printBufferInfo(buffer);
        //put的方法会修改position的值
        buffer.put('H');
        buffer.put('E');
        buffer.put('L');
        buffer.put('L');
        buffer.put('O');
        buffer.limit(8);//修改limit的值
        System.out.println("调用put和limit方法后：");
        printBufferInfo(buffer);
        System.out.println("调用compact方法后：");
        //将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。
        buffer.compact();
        printBufferInfo(buffer);
    }
这里写图片描述

这是JDK中介绍该方法的作用：

将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。即将索引 p = position() 处的字节复制到索引 0 处，将索引 p + 1 处的字节复制到索引 1 处，依此类推，直到将索引 limit() - 1 处的字节复制到索引 n = limit() - 1 - p 处。然后将缓冲区的位置设置为 n+1，并将其界限设置为其容量。如果已定义了标记，则丢弃它。

官方表示的太难理解了：

将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。并将limit(实际容量)设置为 capacity(最大容量)。执行compact()方法前，limit的值是：8，position的值是：5。按照上面描述的执行完compact()后，position的值计算方式是：n+1;n=limit-1-p;所有n=8-1-5=2,最后position的值为：2+1=3。和程序运行的结果一致。
可以在这种情况：从缓冲区写入数据之后调用此方法，以防写入不完整。

buf.clear();          // Prepare buffer for use
  while (in.read(buf) >= 0 || buf.position != 0) {
     buf.flip();
     out.write(buf);
     buf.compact();    // In case of partial write
 }

如果out.write()方法没有将缓存中的数据读取完，这个时候的position位置指向的是剩余数据的位置。达到防止写入不完整。

通道

作用： 连接缓冲区与数据源或数据目的地。

常用类：

Channel
接口有下面两个子接口ReadableByteChannel和WritableByteChannel和一个抽象实现类SelectableChannel。
在ReadableByteChannel接口中申明了read(ByteBuffer
dst)方法。在WritableByteChannel接口中申明了write(ByteBuffer[]
srcs)：方法。SelectableChannel抽象类中主要方法，configureBlocking(boolean
block)、register();方法。 ByteChannel
接口继承了ReadableChannel和WritableChannel。所以ByteChannel具有读和写的功能。

ServerSocketChannel继承了SelectableChannel类抽象类，所以SocketChannel具有设置是否是阻塞模式、向selector注册事件功能。

SocketChannel也继承了SelectableChannel类还实现ByteChannel接口，所以SocketChannel具有设置是否是阻塞模式、向selector注册事件、从缓冲区读写数据的功能。

通过类图展现：


这里写图片描述

Selector类：

作用：只要ServerSocketChannel及SocketChannel向Selector注册了特定的事件，Selector就会监听这些事件的发生。

流程：
Selector通过静态的open()方法创建一个Selector对象，SelectableChannel类向Selector注册了特定的事件。Selector就会监控这些事件发生，Selector通过select()监控已发生的SelectionKey对象的数目，通过selectKeys()方法返回对应的selectionKey对象集合。遍历该集合得到相应的selectionKey对象，通过该对象的channel()方法获取关联的SelectableChannel对象，
通过selector()方法就可以获取关联的Selector对象。

<font color='red'><b>Note：</b></font>
当Selector的select()方法还有一个重载方式:select(long timeout)。并且该方法采用阻塞的工作方式，如果相关事件的selectionKey对象的数目一个也没有，就进入阻塞状态。知道出现以下情况之一，才从select()方法中返回。

  • 至少有一个SelectionKey的相关事件已经发生。
  • 其他线程调用了Selector的wakeup()方法，导致执行select()方法的线程立即返回。
  • 当前执行的select()方法的线程被中断。
  • 超出了等待时间。仅限调用select(long timeout)方法时出现。如果没有设置超时时间，则永远不会超时。

Selector类有两个非常重要的方法： 静态方法open()，这是Selector的静态工厂方法，创建一个Selector对象。
selectedKeys()方法返回被Selector捕获的SelectionKey的集合。

SelectionKey类

作用：
ServerSocketChannel或SocketChannel通过register()方法向Selector注册事件时，register()方法会创建一个SelectionKey对象，该对象是用来跟踪注册事件的句柄。在SelectionKey对象的有效期间，Selector会一直监控与SelectionKey对象相关的事件，如果事件发生，就会把SelectionKey对象添加到Selected-keys集合中。

SelectionKey中定义的事件： 定义了4种事件：
1、SelectionKey.OP_ACCEPT：接收连接就绪事件，表示服务器监听到了客户连接，服务器可以接收这个连接了。常量值为16.
2、SelectionKey.OP_CONNECT：连接就绪事件，表示客户与服务器的连接已经建立成功。常量值为8.
3、SelectionKey.OP_READ：读就绪事件，表示通道中已经有了可读数据可以执行读操作。常量值为1.
4、SelectionKey.OP_WRITE：写就绪事件，表示已经可以向通道写数据了。常量值为4.

常用方法：
channel()方法：返回与它关联的SelectedChannel(包括ServerSocketChannel和SocketChannel)。
selector()方法：返回与它关联的Selector对象。
它们之间的关系如下：


这里写图片描述

<a href="https://shop113396382.taobao.com">视频和电子版书籍</a>

欢迎关注微信公众号 在路上的coder 每天分享优秀的Java技术文章！
扫描二维码关注：

这里写图片描述

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,968评论 6赞 482
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,601评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 153,220评论 0赞 344
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,416评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,425评论 5赞 374
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,144评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,432评论 3赞 401
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,088评论 0赞 261
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,586评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,028评论 2赞 325
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,137评论 1赞 334
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,783评论 4赞 324
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,343评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,333评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,559评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,595评论 2赞 355
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,901评论 2赞 345

推荐阅读更多精彩内容