搜索

NIO之多线程协作处理数据读写

源码学徒

共 8346字，需浏览 17分钟

 · 2021-07-04

有道无术，术尚可求也！有术无道，止于术！

经过前面几章的学习，我们已经 能够掌握了JDK NIO的开发方式，我们来总结一下NIO开发的流程：

  1. 创建一个服务端通道 ServerSocketChannel
  2. 创建一个选择器 Selector
  3. 将服务端通道注册到选择器上，并且关注我们感兴趣的事件serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  4. 绑定服务管道的地址 serverSocketChannel.bind(new InetSocketAddress(8989));
  5. 开始进行事件选择，选择我们感兴趣的事件做对应的操作！

具体的代码信息请参照第一章：多路复用模型章节，这里不做太多的赘述！

有关多路复用的概念，我们也在第一章进行了分析。多路复用模型能够最大限度的将一个线程的执行能力榨干，一条线程执行所有的数据，包括新连接的接入、数据的读取、计算与回写，但是假设，我们的数据计算及其缓慢，那么该任务的执行就势必影响下一个新链接的接入！

传统NIO单线程模型

单线程的NIO模型

如图，我们能了解到，单线程情况下，读事件因为要做一些业务性操作（数据库连接、图片、文件下载）等操作，导致线程阻塞再，读事件的处理上，此时单线程程序无法进行下一次新链接的处理！我们对该线程模型进行优化，select事件处理封装为任务，提交到线程池！

NIO多线程模型

image-20210416170643079

上面的这种数据结构能够解决掉因为计算任务耗时过长，导致新链接接入阻塞的问题，我们能否再次进行一次优化呢？

我们能否创建多个事件选择器，每个事件选择器，负责不同的Socket连接，就像下面这种：

NIO多线程优化模型

image-20210416123153046

这样我们就可以每一个Select选择器负责多个客户端Socket连接，主线程只需要将客户端新连接选择一个选择器注册到select选择器上就可以了！所以我们的架构图，就变成了下图这样：

image-20210416134148671

我们在select选择器内部处理计算任务的时候，也可以将任务封装为task，提交到线程池里面去，彻底将新连接接入和读写事件处理分离开，互不影响！事实上，这也是Netty的核心思想之一，我们可以根据上面的图例，自己简单写一个：

代码实现

构建一个事件执行器 对应上图的select选择器

/**
 * Nio事件处理器
 *
 * @author huangfu
 * @date
 */
public class MyNioEventLoop implements Runnable {
    static final ByteBuffer ALLOCATE = ByteBuffer.allocate(128);
    private final Selector selector;
    private final LinkedBlockingQueue<Runnable> linkedBlockingQueue;
    public MyNioEventLoop(Selector selector) {
        this.selector = selector;
        linkedBlockingQueue = new LinkedBlockingQueue<>();
    }

    public Selector getSelector() {
        return selector;
    }

    public LinkedBlockingQueue<Runnable> getLinkedBlockingQueue() {
        return linkedBlockingQueue;
    }

    //忽略  hashCode和eques

    /**
     * 任务处理器
     */
    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                //进行事件选择  这里我们只处理读事件
                if (selector.select() > 0) {
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    //处理读事件
                    while (iterator.hasNext()) {
                        SelectionKey next = iterator.next();
                        iterator.remove();
                        if (next.isReadable()) {
                            SocketChannel channel = (SocketChannel) next.channel();
                            int read = channel.read(ALLOCATE);
                            if(read > 0) {
                                System.out.printf("线程%s【%s】发来消-息:",Thread.currentThread().getName(), channel.getRemoteAddress());
                                System.out.println(new String(ALLOCATE.array(), StandardCharsets.UTF_8));
                            }else if(read == -1) {
                                System.out.println("连接断开");
                                channel.close();
                            }
                            ALLOCATE.clear();
                        }
                    }
                    selectionKeys.clear();
                }else {
                    //处理异步任务  进行注册
                    while (!linkedBlockingQueue.isEmpty()) {
                        Runnable take = linkedBlockingQueue.take();
                        //异步事件执行
                        take.run();
                    }
                }
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

构建一个选择器组

/**
 * 选择器组
 *
 * @author huangfu
 * @date 2021年3月12日09:44:37
 */
public class SelectorGroup {
    private final List<MyNioEventLoop> SELECTOR_GROUP = new ArrayList<>(8);
    private static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
    private final static AtomicInteger IDX = new AtomicInteger();

    /**
     * 初始化选择器
     * @param count 处理器数量
     * @throws IOException 异常欣喜
     */
    public SelectorGroup(int count) throws IOException {

        for (int i = 0; i < count; i++) {
            Selector open = Selector.open();
            MyNioEventLoop myNioEventLoop = new MyNioEventLoop(open);
            SELECTOR_GROUP.add(myNioEventLoop);
        }
    }

    public SelectorGroup() throws IOException {
        this(AVAILABLE_PROCESSORS << 1);
    }

    /**
     * 轮询获取一个选择器
     * @return 返回一个选择器
     */
    public MyNioEventLoop next(){
        int andIncrement = IDX.getAndIncrement();
        int length = SELECTOR_GROUP.size();

        return SELECTOR_GROUP.get(Math.abs(andIncrement % length));
    }
}

构建一个执行器记录器

/**
 * @author huangfu
 * @date
 */
public class ThreadContext {
    /**
     * 记录当前使用过的选择器
     */
    public static final Set<MyNioEventLoop> RUN_SELECT = new HashSet<>();
}

构建一个新连接接入选择器

/**
 * 连接器
 *
 * @author huangfu
 * @date 2021年3月12日10:15:37
 */
public class Acceptor implements Runnable {
    private final ServerSocketChannel serverSocketChannel;
    private final SelectorGroup selectorGroup;

    public Acceptor(ServerSocketChannel serverSocketChannel, SelectorGroup selectorGroup) {
        this.serverSocketChannel = serverSocketChannel;
        this.selectorGroup = selectorGroup;
    }


    @Override
    public void run() {
        try {
            SocketChannel socketChannel = serverSocketChannel.accept();
            MyNioEventLoop next = selectorGroup.next();

            //向队列追加一个注册任务
            next.getLinkedBlockingQueue().offer(() -> {
                try {
                    //客户端注册为非阻塞
                    socketChannel.configureBlocking(false);
                    //注册到选择器 关注一个读事件
                    socketChannel.register(next.getSelector(), SelectionKey.OP_READ);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            //唤醒对应的任务，让其处理异步任务
            next.getSelector().wakeup();


            System.out.println("检测到连接：" + socketChannel.getRemoteAddress());
            //当当前选择器已经被使用过了  就不再使用了，直接注册就行了
            if (ThreadContext.RUN_SELECT.add(next)) {
                //启动任务
                new Thread(next).start();
            }


        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

创建启动器

/**
 * 反应器
 *
 * @author huangfu
 * @date 2021年3月12日10:15:14
 */
public class Reactor implements Runnable {
    private final Selector selector;

    public Reactor(Selector selector) {
        this.selector = selector;
    }

    @Override
    public void run() {
        try {
            System.out.println("服务启动成功");
            while (!Thread.currentThread().isInterrupted()) {
                //d等待连接事件
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey next = iterator.next();
                    iterator.remove();
                    //进行数据分发
                    dispatch(next);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 将新连接分发到新连接接入器
     * @param next 所有事件主键
     */
    private void dispatch(SelectionKey next) {
        Runnable attachment = (Runnable) next.attachment();
        if(attachment!=null) {
            attachment.run();
        }
    }
}

启动测试

/**
 * @author huangfu
 * @date
 */
public class TestMain {

    public static void main(String[] args) throws IOException {
        //创建一个选择器组   传递选择器组的大小 决定使用多少选择器来实现
        SelectorGroup selectorGroup = new SelectorGroup(2);
        //开启一个服务端管道
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //开启一个服务端专用的选择器
        Selector selector = Selector.open();
        //设置非阻塞
        serverSocketChannel.configureBlocking(false);
        //创建一个连接器
        Acceptor acceptor = new Acceptor(serverSocketChannel, selectorGroup);
        //将服务端通道注册到服务端选择器上  这里会绑定一个新连接接入器
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, acceptor);
        //绑定端口
        serverSocketChannel.bind(new InetSocketAddress(8989));
        //启动处理器
        new Reactor(selector).run();
    }
}

总结

  1. 单线程下的NIO存在性能瓶颈，当某一计算过程缓慢的时候会阻塞住整个线程，导致影响其他事件的处理！
  2. 为了解决这一缺陷，我们提出了使用异步线程的方式去操作任务，将耗时较长的业务，封装为一个异步任务，提交到线程池执行！
  3. 为了使业务操作和新连接接入完全分离开，我们做了另外一重优化，我们封装了一个选择器组，轮询的方式获取选择器，每一个选择器都能够处理多个新连接， socket连接->selector选择器 = 多 -> 1，在每一个选择器里面又可以使用线程池来处理任务，进一步提高吞吐量！

才疏学浅，如果文章中理解有误，欢迎大佬们私聊指正！欢迎关注作者的公众号，一起进步，一起学习！



❤️「转发」和「在看」，是对我最大的支持❤️



浏览 16
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
小美播报|3月IPTV数据排行榜发布！
小美播报3月IPTV数据排行榜:《与凤行》登顶连续剧榜榜首拥有4.05亿家庭用户的中国IPTV平台已经成为国内主流视听平台，IPTV平台数据对视听产业各环节都具有重要意义。截至2024年4月，全国已有29个省级IPTV加入“看中国”，覆盖全国超2.25亿户家庭、辐射近7亿人。点击查看详情湖南广电与马
流媒体网
0
图像处理基础知识
点击上方“小白学视觉”，选择加"星标"或“置顶”重磅干货，第一时间送达图像1、模拟图像模拟图像，又称连续图像，是指在二维坐标系中连续变化的图像，即图像的像点是无限稠密的，同时具有灰度值（即图像从暗到亮的变化值）。2、数字图像数字图像，又称数码图像或数位图像，是二维图像用有限数字数值像素的表示。数字图
小白学视觉
10
python读取一个文件里面几百个csv数据集然后按照列名合并一个数据集
点击上方“Python爬虫与数据挖掘”，进行关注回复“书籍”即可获赠Python从入门到进阶共10本电子书今日鸡汤但使龙城飞将在，不教胡马度阴山。大家好，我是Python进阶者。一、前言前几天在Python最强王者交流群【FiNε_】问了一个Python自动化办公，问题如下：python 读取一个文
Python爬虫与数据挖掘
3
AKShare-股票数据-恐惧贪婪指数
作者寄语本次更新股票数据-恐惧贪婪指数接口。主要修复该接口，目前该接口可以获取 A 股 恐惧贪婪指数 数据。欢迎加入专注于财经数据和量化投资的【数据科学实战】社区，加入方式参见文末！更新接口"index_fear_greed_funddb"  # 恐惧贪婪指数恐惧贪婪指数接口: index
数据科学实战
10
实战必备-数据清洗、数据存储、数据可视化-《AKShare-初阶-实战应用》
✔️课程介绍本课程主要给大家介绍 AKShare 的实战应用部分，主要包括对 AKShare 获取到的数据进行数据清洗，数据存储和数据可视化。课中会提供我们在进行数据处理中的最佳实践！课程目录✔️获取方式如下：  直接购买订阅： ¥39.8/年🔗扫码进入课程店铺：🔅懒得
数据科学实战
31
图像处理，计算机视觉和人工智能之间的差异
作者：Pallawi原文链接：https://medium.com/@pallawi.ds/difference-between-image-processing-computer-vision-and-artificial-intelligence-af670d65055d编译：AI算法与图像处理
机器学习初学者
10
怎么样python把这个excel列数据中双引号去掉呀？
点击上方“Python爬虫与数据挖掘”，进行关注回复“书籍”即可获赠Python从入门到进阶共10本电子书今日鸡汤万籁此都寂，但余钟磬音。大家好，我是皮皮。一、前言前几天在Python最强王者交流群【6G】问了一个Pandas数据分析的问题，这里拿出来给大家分享下。数据截图如下：二、实现过程其实Ex
Python爬虫与数据挖掘
4
面试官：来说说vue3是怎么处理内置的v-for、v-model等指令？
前言最近有粉丝找到我，说被面试官给问懵了。粉丝：面试官上来就问“一个vue文件是如何渲染成浏览器上面的真实DOM？”，当时还挺窃喜这题真简单。就简单说了一下先是编译成render函数、然后根据render函数生成虚拟DOM，最后就是根据虚拟DOM生成真实DOM。按照正常套路面试官接着会问vue响应式
高级前端进阶
10
C# 优雅的处理 TCP 数据
前言Tcp是一个面向连接的流数据传输协议，用人话说就是传输是一个已经建立好连接的管道，数据都在管道里像流水一样流淌到对端。那么数据必然存在几个问题，比如数据如何持续的读取，数据包的边界等。Nagle's算法Nagle 算法的核心思想是，在一个 TCP 连接上，最多只能有一个未被确认的小数据包（小于
dotNET全栈开发
1075
为啥我敢说Python是数据分析界的扛把子语言？
首先声明下这篇文字不是卖课的，也不是无脑吹Python，咱只讲事实，认认真真讨论下Python是不是数据分析领域最好的语言。因为我在知乎上看到非常多人在问这个问题，想必大家是关心的。我的观点是，目前来看所有编程语言里，做数据分析Python是最好的选择，没有之一。列举几个事实：1、Kaggle、天池
Python大数据分析
9
点赞
评论
收藏
分享

手机扫一扫分享

举报